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
The following commit(s) were added to refs/heads/trunk by this push: new 9bde713 Fix IR prepare anti-compaction race 9bde713 is described below commit 9bde713ee8883f70d130efb6290ec0e6daea524f Author: Stefan Podkowinski <s.podkowin...@gmail.com> AuthorDate: Fri Feb 15 10:03:47 2019 +0100 Fix IR prepare anti-compaction race Patch by Stefan Podkowinski; Reviewed by Blake Eggleston for CASSANDRA-15027 --- CHANGES.txt | 1 + .../cassandra/db/compaction/CompactionManager.java | 50 +++- .../db/repair/CassandraKeyspaceRepairManager.java | 6 +- .../cassandra/db/repair/PendingAntiCompaction.java | 27 +- .../cassandra/repair/KeyspaceRepairManager.java | 4 +- .../repair/consistent/CoordinatorSession.java | 94 ++++-- .../cassandra/repair/consistent/LocalSessions.java | 28 +- .../db/compaction/AntiCompactionBytemanTest.java | 2 +- .../db/compaction/AntiCompactionTest.java | 10 +- .../db/compaction/CancelCompactionsTest.java | 2 +- .../repair/PendingAntiCompactionBytemanTest.java | 2 +- .../db/repair/PendingAntiCompactionTest.java | 45 ++- .../apache/cassandra/net/MockMessagingService.java | 8 + .../consistent/CoordinatorMessagingTest.java | 329 +++++++++++++++++++++ .../repair/consistent/CoordinatorSessionTest.java | 31 +- .../repair/consistent/LocalSessionTest.java | 42 ++- 16 files changed, 586 insertions(+), 95 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 02233dd..87d815b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0 + * Fix IR prepare anti-compaction race (CASSANDRA-15027) * Fix SimpleStrategy option validation (CASSANDRA-15007) * Don't try to cancel 2i compactions when starting anticompaction (CASSANDRA-15024) * Avoid NPE in RepairRunnable.recordFailure (CASSANDRA-15025) diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index 85aff08..b388098 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -22,6 +22,7 @@ import java.io.IOException; import java.util.*; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BooleanSupplier; import java.util.function.Predicate; import java.util.stream.Collectors; import javax.management.openmbean.OpenDataException; @@ -654,7 +655,8 @@ public class CompactionManager implements CompactionManagerMBean RangesAtEndpoint tokenRanges, Refs<SSTableReader> sstables, LifecycleTransaction txn, - UUID sessionId) + UUID sessionId, + BooleanSupplier isCancelled) { Runnable runnable = new WrappedRunnable() { @@ -662,7 +664,7 @@ public class CompactionManager implements CompactionManagerMBean { try (TableMetrics.TableTimer.Context ctx = cfs.metric.anticompactionTime.time()) { - performAnticompaction(cfs, tokenRanges, sstables, txn, sessionId); + performAnticompaction(cfs, tokenRanges, sstables, txn, sessionId, isCancelled); } } }; @@ -719,6 +721,7 @@ public class CompactionManager implements CompactionManagerMBean * @param replicas token ranges to be repaired * @param validatedForRepair SSTables containing the repaired ranges. Should be referenced before passing them. * @param sessionID the repair session we're anti-compacting for + * @param isCancelled function that indicates if active anti-compaction should be canceled * @throws InterruptedException * @throws IOException */ @@ -726,7 +729,8 @@ public class CompactionManager implements CompactionManagerMBean RangesAtEndpoint replicas, Refs<SSTableReader> validatedForRepair, LifecycleTransaction txn, - UUID sessionID) throws IOException + UUID sessionID, + BooleanSupplier isCancelled) throws IOException { try { @@ -746,7 +750,7 @@ public class CompactionManager implements CompactionManagerMBean assert txn.originals().equals(sstables); if (!sstables.isEmpty()) - doAntiCompaction(cfs, replicas, txn, sessionID); + doAntiCompaction(cfs, replicas, txn, sessionID, isCancelled); txn.finish(); } finally @@ -1432,11 +1436,14 @@ public class CompactionManager implements CompactionManagerMBean * @param txn a transaction over the repaired sstables to anticompact * @param ranges full and transient ranges to be placed into one of the new sstables. The repaired table will be tracked via * the {@link org.apache.cassandra.io.sstable.metadata.StatsMetadata#pendingRepair} field. + * @param sessionID the repair session we're anti-compacting for + * @param isCancelled function that indicates if active anti-compaction should be canceled */ private void doAntiCompaction(ColumnFamilyStore cfs, RangesAtEndpoint ranges, LifecycleTransaction txn, - UUID pendingRepair) + UUID pendingRepair, + BooleanSupplier isCancelled) { int originalCount = txn.originals().size(); logger.info("Performing anticompaction on {} sstables", originalCount); @@ -1458,7 +1465,7 @@ public class CompactionManager implements CompactionManagerMBean { try (LifecycleTransaction groupTxn = txn.split(sstableGroup)) { - int antiCompacted = antiCompactGroup(cfs, ranges, groupTxn, pendingRepair); + int antiCompacted = antiCompactGroup(cfs, ranges, groupTxn, pendingRepair, isCancelled); antiCompactedSSTableCount += antiCompacted; } } @@ -1469,9 +1476,10 @@ public class CompactionManager implements CompactionManagerMBean @VisibleForTesting int antiCompactGroup(ColumnFamilyStore cfs, - RangesAtEndpoint ranges, - LifecycleTransaction txn, - UUID pendingRepair) + RangesAtEndpoint ranges, + LifecycleTransaction txn, + UUID pendingRepair, + BooleanSupplier isCancelled) { Preconditions.checkArgument(!ranges.isEmpty(), "need at least one full or transient range"); long groupMaxDataAge = -1; @@ -1538,7 +1546,7 @@ public class CompactionManager implements CompactionManagerMBean AbstractCompactionStrategy.ScannerList scanners = strategy.getScanners(txn.originals()); CompactionController controller = new CompactionController(cfs, sstableAsSet, getDefaultGcBefore(cfs, nowInSec)); - CompactionIterator ci = getAntiCompactionIterator(scanners.scanners, controller, nowInSec, UUIDGen.getTimeUUID(), active)) + CompactionIterator ci = getAntiCompactionIterator(scanners.scanners, controller, nowInSec, UUIDGen.getTimeUUID(), active, isCancelled)) { int expectedBloomFilterSize = Math.max(cfs.metadata().params.minIndexInterval, (int)(SSTableReader.getApproximateKeyCount(sstableAsSet))); @@ -1601,16 +1609,30 @@ public class CompactionManager implements CompactionManagerMBean } catch (Throwable e) { - JVMStabilityInspector.inspectThrowable(e); - logger.error("Error anticompacting " + txn, e); + if (e instanceof CompactionInterruptedException && isCancelled.getAsBoolean()) + { + logger.info("Anticompaction has been canceled for session {}", pendingRepair); + logger.trace(e.getMessage(), e); + } + else + { + JVMStabilityInspector.inspectThrowable(e); + logger.error("Error anticompacting " + txn + " for " + pendingRepair, e); + } throw e; } } @VisibleForTesting - public static CompactionIterator getAntiCompactionIterator(List<ISSTableScanner> scanners, CompactionController controller, int nowInSec, UUID timeUUID, ActiveCompactionsTracker activeCompactions) + public static CompactionIterator getAntiCompactionIterator(List<ISSTableScanner> scanners, CompactionController controller, int nowInSec, UUID timeUUID, ActiveCompactionsTracker activeCompactions, BooleanSupplier isCancelled) { - return new CompactionIterator(OperationType.ANTICOMPACTION, scanners, controller, nowInSec, timeUUID, activeCompactions); + return new CompactionIterator(OperationType.ANTICOMPACTION, scanners, controller, nowInSec, timeUUID, activeCompactions) { + + public boolean isStopRequested() + { + return super.isStopRequested() || isCancelled.getAsBoolean(); + } + }; } @VisibleForTesting diff --git a/src/java/org/apache/cassandra/db/repair/CassandraKeyspaceRepairManager.java b/src/java/org/apache/cassandra/db/repair/CassandraKeyspaceRepairManager.java index fa2e653..4fa8650 100644 --- a/src/java/org/apache/cassandra/db/repair/CassandraKeyspaceRepairManager.java +++ b/src/java/org/apache/cassandra/db/repair/CassandraKeyspaceRepairManager.java @@ -21,6 +21,7 @@ package org.apache.cassandra.db.repair; import java.util.Collection; import java.util.UUID; import java.util.concurrent.ExecutorService; +import java.util.function.BooleanSupplier; import com.google.common.util.concurrent.ListenableFuture; @@ -42,9 +43,10 @@ public class CassandraKeyspaceRepairManager implements KeyspaceRepairManager public ListenableFuture prepareIncrementalRepair(UUID sessionID, Collection<ColumnFamilyStore> tables, RangesAtEndpoint tokenRanges, - ExecutorService executor) + ExecutorService executor, + BooleanSupplier isCancelled) { - PendingAntiCompaction pac = new PendingAntiCompaction(sessionID, tables, tokenRanges, executor); + PendingAntiCompaction pac = new PendingAntiCompaction(sessionID, tables, tokenRanges, executor, isCancelled); return pac.run(); } } diff --git a/src/java/org/apache/cassandra/db/repair/PendingAntiCompaction.java b/src/java/org/apache/cassandra/db/repair/PendingAntiCompaction.java index d5c3ca0..fac164d 100644 --- a/src/java/org/apache/cassandra/db/repair/PendingAntiCompaction.java +++ b/src/java/org/apache/cassandra/db/repair/PendingAntiCompaction.java @@ -26,6 +26,7 @@ import java.util.UUID; import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; +import java.util.function.BooleanSupplier; import java.util.stream.Collectors; import com.google.common.annotations.VisibleForTesting; @@ -243,16 +244,18 @@ public class PendingAntiCompaction { private final UUID parentRepairSession; private final RangesAtEndpoint tokenRanges; + private final BooleanSupplier isCancelled; - public AcquisitionCallback(UUID parentRepairSession, RangesAtEndpoint tokenRanges) + public AcquisitionCallback(UUID parentRepairSession, RangesAtEndpoint tokenRanges, BooleanSupplier isCancelled) { this.parentRepairSession = parentRepairSession; this.tokenRanges = tokenRanges; + this.isCancelled = isCancelled; } ListenableFuture<?> submitPendingAntiCompaction(AcquireResult result) { - return CompactionManager.instance.submitPendingAntiCompaction(result.cfs, tokenRanges, result.refs, result.txn, parentRepairSession); + return CompactionManager.instance.submitPendingAntiCompaction(result.cfs, tokenRanges, result.refs, result.txn, parentRepairSession, isCancelled); } private static boolean shouldAbort(AcquireResult result) @@ -312,22 +315,25 @@ public class PendingAntiCompaction private final ExecutorService executor; private final int acquireRetrySeconds; private final int acquireSleepMillis; + private final BooleanSupplier isCancelled; public PendingAntiCompaction(UUID prsId, Collection<ColumnFamilyStore> tables, RangesAtEndpoint tokenRanges, - ExecutorService executor) + ExecutorService executor, + BooleanSupplier isCancelled) { - this(prsId, tables, tokenRanges, ACQUIRE_RETRY_SECONDS, ACQUIRE_SLEEP_MS, executor); + this(prsId, tables, tokenRanges, ACQUIRE_RETRY_SECONDS, ACQUIRE_SLEEP_MS, executor, isCancelled); } @VisibleForTesting PendingAntiCompaction(UUID prsId, - Collection<ColumnFamilyStore> tables, - RangesAtEndpoint tokenRanges, - int acquireRetrySeconds, - int acquireSleepMillis, - ExecutorService executor) + Collection<ColumnFamilyStore> tables, + RangesAtEndpoint tokenRanges, + int acquireRetrySeconds, + int acquireSleepMillis, + ExecutorService executor, + BooleanSupplier isCancelled) { this.prsId = prsId; this.tables = tables; @@ -335,6 +341,7 @@ public class PendingAntiCompaction this.executor = executor; this.acquireRetrySeconds = acquireRetrySeconds; this.acquireSleepMillis = acquireSleepMillis; + this.isCancelled = isCancelled; } public ListenableFuture run() @@ -361,6 +368,6 @@ public class PendingAntiCompaction @VisibleForTesting protected AcquisitionCallback getAcquisitionCallback(UUID prsId, RangesAtEndpoint tokenRanges) { - return new AcquisitionCallback(prsId, tokenRanges); + return new AcquisitionCallback(prsId, tokenRanges, isCancelled); } } diff --git a/src/java/org/apache/cassandra/repair/KeyspaceRepairManager.java b/src/java/org/apache/cassandra/repair/KeyspaceRepairManager.java index bc614dc..0739f10 100644 --- a/src/java/org/apache/cassandra/repair/KeyspaceRepairManager.java +++ b/src/java/org/apache/cassandra/repair/KeyspaceRepairManager.java @@ -21,6 +21,7 @@ package org.apache.cassandra.repair; import java.util.Collection; import java.util.UUID; import java.util.concurrent.ExecutorService; +import java.util.function.BooleanSupplier; import com.google.common.util.concurrent.ListenableFuture; @@ -40,5 +41,6 @@ public interface KeyspaceRepairManager ListenableFuture prepareIncrementalRepair(UUID sessionID, Collection<ColumnFamilyStore> tables, RangesAtEndpoint tokenRanges, - ExecutorService executor); + ExecutorService executor, + BooleanSupplier isCancelled); } diff --git a/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java b/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java index f52a28d..b921342 100644 --- a/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java +++ b/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java @@ -26,6 +26,7 @@ import java.util.function.Supplier; import javax.annotation.Nullable; +import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import com.google.common.collect.Iterables; import com.google.common.util.concurrent.AsyncFunction; @@ -95,6 +96,12 @@ public class CoordinatorSession extends ConsistentSession super.setState(state); } + @VisibleForTesting + synchronized State getParticipantState(InetAddressAndPort participant) + { + return participantStates.get(participant); + } + public synchronized void setParticipantState(InetAddressAndPort participant, State state) { logger.trace("Setting participant {} to state {} for repair {}", participant, state, sessionID); @@ -153,25 +160,31 @@ public class CoordinatorSession extends ConsistentSession public synchronized void handlePrepareResponse(InetAddressAndPort participant, boolean success) { - if (getState() == State.FAILED) - { - logger.trace("Incremental repair session {} has failed, ignoring prepare response from {}", sessionID, participant); - } - else if (!success) + if (!success) { - logger.debug("{} failed the prepare phase for incremental repair session {}. Aborting session", participant, sessionID); - fail(); - prepareFuture.set(false); + logger.debug("{} failed the prepare phase for incremental repair session {}", participant, sessionID); + sendFailureMessageToParticipants(); + setParticipantState(participant, State.FAILED); } else { logger.trace("Successful prepare response received from {} for repair session {}", participant, sessionID); setParticipantState(participant, State.PREPARED); - if (getState() == State.PREPARED) - { - logger.debug("Incremental repair session {} successfully prepared.", sessionID); - prepareFuture.set(true); - } + } + + // don't progress until we've heard from all replicas + if(Iterables.any(participantStates.values(), v -> v == State.PREPARING)) + return; + + if (getState() == State.PREPARED) + { + logger.info("Incremental repair session {} successfully prepared.", sessionID); + prepareFuture.set(true); + } + else + { + fail(); + prepareFuture.set(false); } } @@ -229,9 +242,8 @@ public class CoordinatorSession extends ConsistentSession logger.info("Incremental repair session {} completed", sessionID); } - public synchronized void fail() + private void sendFailureMessageToParticipants() { - logger.info("Incremental repair session {} failed", sessionID); FailSession message = new FailSession(sessionID); for (final InetAddressAndPort participant : participants) { @@ -240,6 +252,12 @@ public class CoordinatorSession extends ConsistentSession sendMessage(participant, message); } } + } + + public synchronized void fail() + { + logger.info("Incremental repair session {} failed", sessionID); + sendFailureMessageToParticipants(); setAll(State.FAILED); String exceptionMsg = String.format("Incremental repair session %s has failed", sessionID); @@ -312,41 +330,59 @@ public class CoordinatorSession extends ConsistentSession } }, MoreExecutors.directExecutor()); + // return execution result as set by following callback + SettableFuture<Boolean> resultFuture = SettableFuture.create(); + // commit repaired data Futures.addCallback(proposeFuture, new FutureCallback<Boolean>() { public void onSuccess(@Nullable Boolean result) { - if (result != null && result) + try { - if (logger.isDebugEnabled()) + if (result != null && result) { - logger.debug("Incremental repair {} finalization phase completed in {}", sessionID, formatDuration(finalizeStart, System.currentTimeMillis())); + if (logger.isDebugEnabled()) + { + logger.debug("Incremental repair {} finalization phase completed in {}", sessionID, formatDuration(finalizeStart, System.currentTimeMillis())); + } + finalizeCommit(); + if (logger.isDebugEnabled()) + { + logger.debug("Incremental repair {} phase completed in {}", sessionID, formatDuration(sessionStart, System.currentTimeMillis())); + } } - finalizeCommit(); - if (logger.isDebugEnabled()) + else { - logger.debug("Incremental repair {} phase completed in {}", sessionID, formatDuration(sessionStart, System.currentTimeMillis())); + hasFailure.set(true); + fail(); } + resultFuture.set(result); } - else + catch (Exception e) { - hasFailure.set(true); - fail(); + resultFuture.setException(e); } } public void onFailure(Throwable t) { - if (logger.isDebugEnabled()) + try + { + if (logger.isDebugEnabled()) + { + logger.debug("Incremental repair {} phase failed in {}", sessionID, formatDuration(sessionStart, System.currentTimeMillis())); + } + hasFailure.set(true); + fail(); + } + finally { - logger.debug("Incremental repair {} phase failed in {}", sessionID, formatDuration(sessionStart, System.currentTimeMillis())); + resultFuture.setException(t); } - hasFailure.set(true); - fail(); } }); - return proposeFuture; + return resultFuture; } } diff --git a/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java b/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java index c28391a..c39c4e6 100644 --- a/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java +++ b/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java @@ -33,6 +33,7 @@ import java.util.UUID; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import java.util.function.BooleanSupplier; import java.util.function.Predicate; import java.util.stream.Collectors; import javax.annotation.Nullable; @@ -523,11 +524,17 @@ public class LocalSessions public void failSession(UUID sessionID, boolean sendMessage) { - logger.info("Failing local repair session {}", sessionID); LocalSession session = getSession(sessionID); if (session != null) { - setStateAndSave(session, FAILED); + synchronized (session) + { + if (session.getState() != FAILED) + { + logger.info("Failing local repair session {}", sessionID); + setStateAndSave(session, FAILED); + } + } if (sendMessage) { sendMessage(session.coordinator, new FailSession(sessionID)); @@ -550,9 +557,10 @@ public class LocalSessions UUID sessionID, Collection<ColumnFamilyStore> tables, RangesAtEndpoint tokenRanges, - ExecutorService executor) + ExecutorService executor, + BooleanSupplier isCancelled) { - return repairManager.prepareIncrementalRepair(sessionID, tables, tokenRanges, executor); + return repairManager.prepareIncrementalRepair(sessionID, tables, tokenRanges, executor, isCancelled); } RangesAtEndpoint filterLocalRanges(String keyspace, Set<Range<Token>> ranges) @@ -599,8 +607,8 @@ public class LocalSessions } catch (Throwable e) { - logger.trace("Error retrieving ParentRepairSession for session {}, responding with failure", sessionID); - sendMessage(coordinator, new FailSession(sessionID)); + logger.error("Error retrieving ParentRepairSession for session {}, responding with failure", sessionID); + sendMessage(coordinator, new PrepareConsistentResponse(sessionID, getBroadcastAddressAndPort(), false)); return; } @@ -612,7 +620,8 @@ public class LocalSessions KeyspaceRepairManager repairManager = parentSession.getKeyspace().getRepairManager(); RangesAtEndpoint tokenRanges = filterLocalRanges(parentSession.getKeyspace().getName(), parentSession.getRanges()); - ListenableFuture repairPreparation = prepareSession(repairManager, sessionID, parentSession.getColumnFamilyStores(), tokenRanges, executor); + ListenableFuture repairPreparation = prepareSession(repairManager, sessionID, parentSession.getColumnFamilyStores(), + tokenRanges, executor, () -> session.getState() != PREPARING); Futures.addCallback(repairPreparation, new FutureCallback<Object>() { @@ -620,7 +629,7 @@ public class LocalSessions { try { - logger.debug("Prepare phase for incremental repair session {} completed", sessionID); + logger.info("Prepare phase for incremental repair session {} completed", sessionID); if (session.getState() != FAILED) { setStateAndSave(session, PREPARED); @@ -628,7 +637,8 @@ public class LocalSessions } else { - logger.debug("Session {} failed before anticompaction completed", sessionID); + logger.info("Session {} failed before anticompaction completed", sessionID); + sendMessage(coordinator, new PrepareConsistentResponse(sessionID, getBroadcastAddressAndPort(), false)); } } finally diff --git a/test/unit/org/apache/cassandra/db/compaction/AntiCompactionBytemanTest.java b/test/unit/org/apache/cassandra/db/compaction/AntiCompactionBytemanTest.java index 31187c7..1673d01 100644 --- a/test/unit/org/apache/cassandra/db/compaction/AntiCompactionBytemanTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/AntiCompactionBytemanTest.java @@ -125,7 +125,7 @@ public class AntiCompactionBytemanTest extends CQLTester try (LifecycleTransaction txn = getCurrentColumnFamilyStore().getTracker().tryModify(getCurrentColumnFamilyStore().getLiveSSTables(), OperationType.ANTICOMPACTION)) { - CompactionManager.instance.antiCompactGroup(getCurrentColumnFamilyStore(), ranges, txn, UUID.randomUUID()); + CompactionManager.instance.antiCompactGroup(getCurrentColumnFamilyStore(), ranges, txn, UUID.randomUUID(), () -> false); } finished.set(true); t.join(); diff --git a/test/unit/org/apache/cassandra/db/compaction/AntiCompactionTest.java b/test/unit/org/apache/cassandra/db/compaction/AntiCompactionTest.java index 08609f3..7b5164a 100644 --- a/test/unit/org/apache/cassandra/db/compaction/AntiCompactionTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/AntiCompactionTest.java @@ -150,7 +150,7 @@ public class AntiCompactionTest if (txn == null) throw new IllegalStateException(); registerParentRepairSession(sessionID, ranges.ranges(), FBUtilities.nowInSeconds(), sessionID); - CompactionManager.instance.performAnticompaction(store, ranges, refs, txn, sessionID); + CompactionManager.instance.performAnticompaction(store, ranges, refs, txn, sessionID, () -> false); } SSTableStats stats = new SSTableStats(); @@ -251,7 +251,7 @@ public class AntiCompactionTest try (LifecycleTransaction txn = cfs.getTracker().tryModify(sstables, OperationType.ANTICOMPACTION); Refs<SSTableReader> refs = Refs.ref(sstables)) { - CompactionManager.instance.performAnticompaction(cfs, atEndpoint(ranges, NO_RANGES), refs, txn, parentRepairSession); + CompactionManager.instance.performAnticompaction(cfs, atEndpoint(ranges, NO_RANGES), refs, txn, parentRepairSession, () -> false); } long sum = 0; long rows = 0; @@ -381,7 +381,7 @@ public class AntiCompactionTest try (LifecycleTransaction txn = store.getTracker().tryModify(sstables, OperationType.ANTICOMPACTION); Refs<SSTableReader> refs = Refs.ref(sstables)) { - CompactionManager.instance.performAnticompaction(store, atEndpoint(ranges, NO_RANGES), refs, txn, pendingRepair); + CompactionManager.instance.performAnticompaction(store, atEndpoint(ranges, NO_RANGES), refs, txn, pendingRepair, () -> false); } assertThat(store.getLiveSSTables().size(), is(1)); @@ -415,7 +415,7 @@ public class AntiCompactionTest try (LifecycleTransaction txn = store.getTracker().tryModify(sstables, OperationType.ANTICOMPACTION); Refs<SSTableReader> refs = Refs.ref(sstables)) { - CompactionManager.instance.performAnticompaction(store, atEndpoint(ranges, NO_RANGES), refs, txn, parentRepairSession); + CompactionManager.instance.performAnticompaction(store, atEndpoint(ranges, NO_RANGES), refs, txn, parentRepairSession, () -> false); } catch (IllegalStateException e) { @@ -485,7 +485,7 @@ public class AntiCompactionTest Assert.assertFalse(refs.isEmpty()); try { - CompactionManager.instance.performAnticompaction(store, atEndpoint(ranges, NO_RANGES), refs, txn, missingRepairSession); + CompactionManager.instance.performAnticompaction(store, atEndpoint(ranges, NO_RANGES), refs, txn, missingRepairSession, () -> false); Assert.fail("expected RuntimeException"); } catch (RuntimeException e) diff --git a/test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java b/test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java index cb4ef4a..976180c 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java @@ -259,7 +259,7 @@ public class CancelCompactionsTest extends CQLTester InetAddressAndPort local = FBUtilities.getBroadcastAddressAndPort(); RangesAtEndpoint rae = RangesAtEndpoint.builder(local).add(new Replica(local, range, true)).build(); - PendingAntiCompaction pac = new PendingAntiCompaction(prsid, Collections.singleton(cfs), rae, Executors.newSingleThreadExecutor()); + PendingAntiCompaction pac = new PendingAntiCompaction(prsid, Collections.singleton(cfs), rae, Executors.newSingleThreadExecutor(), () -> false); Future<?> fut = pac.run(); Thread.sleep(600); List<TestCompactionTask> toAbort = new ArrayList<>(); diff --git a/test/unit/org/apache/cassandra/db/repair/PendingAntiCompactionBytemanTest.java b/test/unit/org/apache/cassandra/db/repair/PendingAntiCompactionBytemanTest.java index 2f2612a..9271866 100644 --- a/test/unit/org/apache/cassandra/db/repair/PendingAntiCompactionBytemanTest.java +++ b/test/unit/org/apache/cassandra/db/repair/PendingAntiCompactionBytemanTest.java @@ -67,7 +67,7 @@ public class PendingAntiCompactionBytemanTest extends AbstractPendingAntiCompact UUID prsid = prepareSession(); try { - PendingAntiCompaction pac = new PendingAntiCompaction(prsid, Lists.newArrayList(cfs, cfs2), atEndpoint(ranges, NO_RANGES), es); + PendingAntiCompaction pac = new PendingAntiCompaction(prsid, Lists.newArrayList(cfs, cfs2), atEndpoint(ranges, NO_RANGES), es, () -> false); pac.run().get(); fail("PAC should throw exception when anticompaction throws exception!"); } diff --git a/test/unit/org/apache/cassandra/db/repair/PendingAntiCompactionTest.java b/test/unit/org/apache/cassandra/db/repair/PendingAntiCompactionTest.java index e44c697..b140813 100644 --- a/test/unit/org/apache/cassandra/db/repair/PendingAntiCompactionTest.java +++ b/test/unit/org/apache/cassandra/db/repair/PendingAntiCompactionTest.java @@ -36,6 +36,7 @@ import java.util.stream.Collectors; import javax.annotation.Nullable; +import com.google.common.collect.Iterables; import com.google.common.collect.Lists; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; @@ -94,7 +95,7 @@ public class PendingAntiCompactionTest extends AbstractPendingAntiCompactionTest { public InstrumentedAcquisitionCallback(UUID parentRepairSession, RangesAtEndpoint ranges) { - super(parentRepairSession, ranges); + super(parentRepairSession, ranges, () -> false); } Set<TableId> submittedCompactions = new HashSet<>(); @@ -142,7 +143,7 @@ public class PendingAntiCompactionTest extends AbstractPendingAntiCompactionTest ExecutorService executor = Executors.newSingleThreadExecutor(); try { - pac = new PendingAntiCompaction(sessionID, tables, atEndpoint(ranges, NO_RANGES), executor); + pac = new PendingAntiCompaction(sessionID, tables, atEndpoint(ranges, NO_RANGES), executor, () -> false); pac.run().get(); } finally @@ -371,10 +372,36 @@ public class PendingAntiCompactionTest extends AbstractPendingAntiCompactionTest true,0, true, PreviewKind.NONE); - CompactionManager.instance.performAnticompaction(result.cfs, atEndpoint(FULL_RANGE, NO_RANGES), result.refs, result.txn, sessionID); + CompactionManager.instance.performAnticompaction(result.cfs, atEndpoint(FULL_RANGE, NO_RANGES), result.refs, result.txn, sessionID, () -> false); } + @Test (expected = CompactionInterruptedException.class) + public void cancelledAntiCompaction() throws Exception + { + cfs.disableAutoCompaction(); + makeSSTables(1); + + PendingAntiCompaction.AcquisitionCallable acquisitionCallable = new PendingAntiCompaction.AcquisitionCallable(cfs, FULL_RANGE, UUIDGen.getTimeUUID(), 0, 0); + PendingAntiCompaction.AcquireResult result = acquisitionCallable.call(); + UUID sessionID = UUIDGen.getTimeUUID(); + ActiveRepairService.instance.registerParentRepairSession(sessionID, + InetAddressAndPort.getByName("127.0.0.1"), + Lists.newArrayList(cfs), + FULL_RANGE, + true,0, + true, + PreviewKind.NONE); + + // attempt to anti-compact the sstable in half + SSTableReader sstable = Iterables.getOnlyElement(cfs.getLiveSSTables()); + Token left = cfs.getPartitioner().midpoint(sstable.first.getToken(), sstable.last.getToken()); + Token right = sstable.last.getToken(); + CompactionManager.instance.performAnticompaction(result.cfs, + atEndpoint(Collections.singleton(new Range<>(left, right)), NO_RANGES), + result.refs, result.txn, sessionID, () -> true); + } + /** * Makes sure that PendingAntiCompaction fails when anticompaction throws exception */ @@ -385,11 +412,11 @@ public class PendingAntiCompactionTest extends AbstractPendingAntiCompactionTest makeSSTables(2); UUID prsid = UUID.randomUUID(); ListeningExecutorService es = MoreExecutors.listeningDecorator(MoreExecutors.newDirectExecutorService()); - PendingAntiCompaction pac = new PendingAntiCompaction(prsid, Collections.singleton(cfs), atEndpoint(FULL_RANGE, NO_RANGES), es) { + PendingAntiCompaction pac = new PendingAntiCompaction(prsid, Collections.singleton(cfs), atEndpoint(FULL_RANGE, NO_RANGES), es, () -> false) { @Override protected AcquisitionCallback getAcquisitionCallback(UUID prsId, RangesAtEndpoint tokenRanges) { - return new AcquisitionCallback(prsid, tokenRanges) + return new AcquisitionCallback(prsid, tokenRanges, () -> false) { @Override ListenableFuture<?> submitPendingAntiCompaction(AcquireResult result) @@ -431,13 +458,13 @@ public class PendingAntiCompactionTest extends AbstractPendingAntiCompactionTest { try (LifecycleTransaction txn = cfs.getTracker().tryModify(sstables, OperationType.ANTICOMPACTION); CompactionController controller = new CompactionController(cfs, sstables, 0); - CompactionIterator ci = CompactionManager.getAntiCompactionIterator(scanners, controller, 0, UUID.randomUUID(), CompactionManager.instance.active)) + CompactionIterator ci = CompactionManager.getAntiCompactionIterator(scanners, controller, 0, UUID.randomUUID(), CompactionManager.instance.active, () -> false)) { // `ci` is our imaginary ongoing anticompaction which makes no progress until after 30s // now we try to start a new AC, which will try to cancel all ongoing compactions CompactionManager.instance.active.beginCompaction(ci); - PendingAntiCompaction pac = new PendingAntiCompaction(prsid, Collections.singleton(cfs), atEndpoint(FULL_RANGE, NO_RANGES), 0, 0, es); + PendingAntiCompaction pac = new PendingAntiCompaction(prsid, Collections.singleton(cfs), atEndpoint(FULL_RANGE, NO_RANGES), 0, 0, es, () -> false); ListenableFuture fut = pac.run(); try { @@ -491,7 +518,7 @@ public class PendingAntiCompactionTest extends AbstractPendingAntiCompactionTest // now we try to start a new AC, which will try to cancel all ongoing compactions CompactionManager.instance.active.beginCompaction(ci); - PendingAntiCompaction pac = new PendingAntiCompaction(prsid, Collections.singleton(cfs), atEndpoint(FULL_RANGE, NO_RANGES), es); + PendingAntiCompaction pac = new PendingAntiCompaction(prsid, Collections.singleton(cfs), atEndpoint(FULL_RANGE, NO_RANGES), es, () -> false); ListenableFuture fut = pac.run(); try { @@ -693,7 +720,7 @@ public class PendingAntiCompactionTest extends AbstractPendingAntiCompactionTest // mark the sstables pending, with a 2i compaction going, which should be untouched; try (LifecycleTransaction txn = idx.getTracker().tryModify(idx.getLiveSSTables(), OperationType.COMPACTION)) { - PendingAntiCompaction pac = new PendingAntiCompaction(prsid, Collections.singleton(cfs2), atEndpoint(FULL_RANGE, NO_RANGES), es); + PendingAntiCompaction pac = new PendingAntiCompaction(prsid, Collections.singleton(cfs2), atEndpoint(FULL_RANGE, NO_RANGES), es, () -> false); pac.run().get(); } // and make sure it succeeded; diff --git a/test/unit/org/apache/cassandra/net/MockMessagingService.java b/test/unit/org/apache/cassandra/net/MockMessagingService.java index 4ea1bf5..79edae8 100644 --- a/test/unit/org/apache/cassandra/net/MockMessagingService.java +++ b/test/unit/org/apache/cassandra/net/MockMessagingService.java @@ -81,6 +81,14 @@ public class MockMessagingService } /** + * Creates a matcher that will indicate if the target address of the outgoing message matches the provided predicate. + */ + public static Matcher<InetAddressAndPort> to(Predicate<InetAddressAndPort> predicate) + { + return (in, to) -> predicate.test(to); + } + + /** * Creates a matcher that will indicate if the verb of the outgoing message equals the * provided value. */ diff --git a/test/unit/org/apache/cassandra/repair/consistent/CoordinatorMessagingTest.java b/test/unit/org/apache/cassandra/repair/consistent/CoordinatorMessagingTest.java new file mode 100644 index 0000000..b532abd --- /dev/null +++ b/test/unit/org/apache/cassandra/repair/consistent/CoordinatorMessagingTest.java @@ -0,0 +1,329 @@ +/* + * 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.consistent; + +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Function; +import java.util.function.Supplier; + +import com.google.common.collect.Lists; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.SettableFuture; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.SchemaLoader; +import org.apache.cassandra.cql3.statements.schema.CreateTableStatement; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.net.MessageIn; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.net.MockMessagingService; +import org.apache.cassandra.net.MockMessagingSpy; +import org.apache.cassandra.repair.AbstractRepairTest; +import org.apache.cassandra.repair.RepairSessionResult; +import org.apache.cassandra.repair.messages.FailSession; +import org.apache.cassandra.repair.messages.FinalizeCommit; +import org.apache.cassandra.repair.messages.FinalizePromise; +import org.apache.cassandra.repair.messages.FinalizePropose; +import org.apache.cassandra.repair.messages.PrepareConsistentRequest; +import org.apache.cassandra.repair.messages.PrepareConsistentResponse; +import org.apache.cassandra.schema.KeyspaceParams; +import org.apache.cassandra.schema.Schema; +import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.service.ActiveRepairService; + +import static org.apache.cassandra.net.MockMessagingService.all; +import static org.apache.cassandra.net.MockMessagingService.payload; +import static org.apache.cassandra.net.MockMessagingService.to; +import static org.apache.cassandra.net.MockMessagingService.verb; +import static org.junit.Assert.fail; + +public class CoordinatorMessagingTest extends AbstractRepairTest +{ + + protected ColumnFamilyStore cfs; + + @BeforeClass + public static void setupClass() + { + SchemaLoader.prepareServer(); + LocalSessionAccessor.startup(); + } + + @Before + public void setup() + { + String ks = "ks_" + System.currentTimeMillis(); + TableMetadata cfm = CreateTableStatement.parse(String.format("CREATE TABLE %s.%s (k INT PRIMARY KEY, v INT)", ks, "tbl"), ks).build(); + SchemaLoader.createKeyspace(ks, KeyspaceParams.simple(1), cfm); + cfs = Schema.instance.getColumnFamilyStoreInstance(cfm.id); + cfs.disableAutoCompaction(); + } + + @After + public void cleanup() + { + MockMessagingService.cleanup(); + } + + @Test + public void testMockedMessagingHappyPath() throws InterruptedException, ExecutionException, TimeoutException + { + + MockMessagingSpy spyPrepare = createPrepareSpy(Collections.emptySet(), Collections.emptySet()); + MockMessagingSpy spyFinalize = createFinalizeSpy(Collections.emptySet(), Collections.emptySet()); + MockMessagingSpy spyCommit = createCommitSpy(); + + UUID uuid = registerSession(cfs, true, true); + CoordinatorSession coordinator = ActiveRepairService.instance.consistent.coordinated.registerSession(uuid, PARTICIPANTS, false); + AtomicBoolean repairSubmitted = new AtomicBoolean(false); + SettableFuture<List<RepairSessionResult>> repairFuture = SettableFuture.create(); + Supplier<ListenableFuture<List<RepairSessionResult>>> sessionSupplier = () -> + { + repairSubmitted.set(true); + return repairFuture; + }; + + // coordinator sends prepare requests to create local session and perform anticompaction + AtomicBoolean hasFailures = new AtomicBoolean(false); + Assert.assertFalse(repairSubmitted.get()); + + // execute repair and start prepare phase + ListenableFuture<Boolean> sessionResult = coordinator.execute(sessionSupplier, hasFailures); + Assert.assertFalse(sessionResult.isDone()); + Assert.assertFalse(hasFailures.get()); + // prepare completed + spyPrepare.interceptMessageOut(3).get(1, TimeUnit.SECONDS); + Assert.assertFalse(sessionResult.isDone()); + Assert.assertFalse(hasFailures.get()); + + // set result from local repair session + repairFuture.set(Lists.newArrayList(createResult(coordinator), createResult(coordinator), createResult(coordinator))); + + // finalize phase + spyFinalize.interceptMessageOut(3).get(1, TimeUnit.SECONDS); + Assert.assertFalse(sessionResult.isDone()); + Assert.assertFalse(hasFailures.get()); + + // commit phase + spyCommit.interceptMessageOut(3).get(1, TimeUnit.SECONDS); + Assert.assertTrue(sessionResult.get()); + Assert.assertFalse(hasFailures.get()); + + // expect no other messages except from intercepted so far + spyPrepare.interceptNoMsg(100, TimeUnit.MILLISECONDS); + spyFinalize.interceptNoMsg(100, TimeUnit.MILLISECONDS); + spyCommit.interceptNoMsg(100, TimeUnit.MILLISECONDS); + + Assert.assertEquals(ConsistentSession.State.FINALIZED, coordinator.getState()); + Assert.assertFalse(ActiveRepairService.instance.consistent.local.isSessionInProgress(uuid)); + } + + + @Test + public void testMockedMessagingPrepareFailureP1() throws InterruptedException, ExecutionException, TimeoutException + { + createPrepareSpy(Collections.singleton(PARTICIPANT1), Collections.emptySet()); + testMockedMessagingPrepareFailure(); + } + + @Test + public void testMockedMessagingPrepareFailureP12() throws InterruptedException, ExecutionException, TimeoutException + { + createPrepareSpy(Lists.newArrayList(PARTICIPANT1, PARTICIPANT2), Collections.emptySet()); + testMockedMessagingPrepareFailure(); + } + + @Test + public void testMockedMessagingPrepareFailureP3() throws InterruptedException, ExecutionException, TimeoutException + { + createPrepareSpy(Collections.singleton(PARTICIPANT3), Collections.emptySet()); + testMockedMessagingPrepareFailure(); + } + + @Test + public void testMockedMessagingPrepareFailureP123() throws InterruptedException, ExecutionException, TimeoutException + { + createPrepareSpy(Lists.newArrayList(PARTICIPANT1, PARTICIPANT2, PARTICIPANT3), Collections.emptySet()); + testMockedMessagingPrepareFailure(); + } + + @Test(expected = TimeoutException.class) + public void testMockedMessagingPrepareFailureWrongSessionId() throws InterruptedException, ExecutionException, TimeoutException + { + createPrepareSpy(Collections.singleton(PARTICIPANT1), Collections.emptySet(), (msgOut) -> UUID.randomUUID()); + testMockedMessagingPrepareFailure(); + } + + private void testMockedMessagingPrepareFailure() throws InterruptedException, ExecutionException, TimeoutException + { + // we expect FailSession messages to all participants + MockMessagingSpy sendFailSessionExpectedSpy = createFailSessionSpy(Lists.newArrayList(PARTICIPANT1, PARTICIPANT2, PARTICIPANT3)); + + UUID uuid = registerSession(cfs, true, true); + CoordinatorSession coordinator = ActiveRepairService.instance.consistent.coordinated.registerSession(uuid, PARTICIPANTS, false); + AtomicBoolean repairSubmitted = new AtomicBoolean(false); + SettableFuture<List<RepairSessionResult>> repairFuture = SettableFuture.create(); + Supplier<ListenableFuture<List<RepairSessionResult>>> sessionSupplier = () -> + { + repairSubmitted.set(true); + return repairFuture; + }; + + // coordinator sends prepare requests to create local session and perform anticompaction + AtomicBoolean proposeFailed = new AtomicBoolean(false); + Assert.assertFalse(repairSubmitted.get()); + + // execute repair and start prepare phase + ListenableFuture<Boolean> sessionResult = coordinator.execute(sessionSupplier, proposeFailed); + Assert.assertFalse(proposeFailed.get()); + // prepare completed + try + { + sessionResult.get(1, TimeUnit.SECONDS); + fail("Completed session without failure after prepare failed"); + } + catch (ExecutionException e) + { + } + sendFailSessionExpectedSpy.interceptMessageOut(3).get(1, TimeUnit.SECONDS); + Assert.assertFalse(repairSubmitted.get()); + Assert.assertTrue(proposeFailed.get()); + Assert.assertEquals(ConsistentSession.State.FAILED, coordinator.getState()); + Assert.assertFalse(ActiveRepairService.instance.consistent.local.isSessionInProgress(uuid)); + } + + @Test + public void testMockedMessagingPrepareTimeout() throws InterruptedException, ExecutionException, TimeoutException + { + MockMessagingSpy spyPrepare = createPrepareSpy(Collections.emptySet(), Collections.singleton(PARTICIPANT3)); + MockMessagingSpy sendFailSessionUnexpectedSpy = createFailSessionSpy(Lists.newArrayList(PARTICIPANT1, PARTICIPANT2, PARTICIPANT3)); + + UUID uuid = registerSession(cfs, true, true); + CoordinatorSession coordinator = ActiveRepairService.instance.consistent.coordinated.registerSession(uuid, PARTICIPANTS, false); + AtomicBoolean repairSubmitted = new AtomicBoolean(false); + SettableFuture<List<RepairSessionResult>> repairFuture = SettableFuture.create(); + Supplier<ListenableFuture<List<RepairSessionResult>>> sessionSupplier = () -> + { + repairSubmitted.set(true); + return repairFuture; + }; + + // coordinator sends prepare requests to create local session and perform anticompaction + AtomicBoolean hasFailures = new AtomicBoolean(false); + Assert.assertFalse(repairSubmitted.get()); + + // execute repair and start prepare phase + ListenableFuture<Boolean> sessionResult = coordinator.execute(sessionSupplier, hasFailures); + try + { + sessionResult.get(1, TimeUnit.SECONDS); + fail("Completed session without failure after prepare failed"); + } + catch (ExecutionException e) + { + fail("Failed session in prepare failed during timeout from participant"); + } + catch (TimeoutException e) + { + // expected + } + // we won't send out any fail session message in case of timeouts + spyPrepare.expectMockedMessageIn(2).get(100, TimeUnit.MILLISECONDS); + sendFailSessionUnexpectedSpy.interceptNoMsg(100, TimeUnit.MILLISECONDS); + Assert.assertFalse(repairSubmitted.get()); + Assert.assertFalse(hasFailures.get()); + Assert.assertEquals(ConsistentSession.State.PREPARING, coordinator.getState()); + Assert.assertFalse(ActiveRepairService.instance.consistent.local.isSessionInProgress(uuid)); + } + + private MockMessagingSpy createPrepareSpy(Collection<InetAddressAndPort> failed, + Collection<InetAddressAndPort> timeout) + { + return createPrepareSpy(failed, timeout, (msgOut) -> msgOut.parentSession); + } + + private MockMessagingSpy createPrepareSpy(Collection<InetAddressAndPort> failed, + Collection<InetAddressAndPort> timeout, + Function<PrepareConsistentRequest, UUID> sessionIdFunc) + { + return MockMessagingService.when( + all(verb(MessagingService.Verb.REPAIR_MESSAGE), + payload((p) -> p instanceof PrepareConsistentRequest)) + ).respond((msgOut, to) -> + { + if(timeout.contains(to)) return null; + else return MessageIn.create(to, + new PrepareConsistentResponse(sessionIdFunc.apply((PrepareConsistentRequest) msgOut.payload), to, !failed.contains(to)), + Collections.emptyMap(), + MessagingService.Verb.REPAIR_MESSAGE, + MessagingService.current_version); + }); + } + + private MockMessagingSpy createFinalizeSpy(Collection<InetAddressAndPort> failed, + Collection<InetAddressAndPort> timeout) + { + return MockMessagingService.when( + all(verb(MessagingService.Verb.REPAIR_MESSAGE), + payload((p) -> p instanceof FinalizePropose)) + ).respond((msgOut, to) -> + { + if(timeout.contains(to)) return null; + else return MessageIn.create(to, + new FinalizePromise(((FinalizePropose) msgOut.payload).sessionID, to, !failed.contains(to)), + Collections.emptyMap(), + MessagingService.Verb.REPAIR_MESSAGE, + MessagingService.current_version); + }); + } + + private MockMessagingSpy createCommitSpy() + { + return MockMessagingService.when( + all(verb(MessagingService.Verb.REPAIR_MESSAGE), + payload((p) -> p instanceof FinalizeCommit)) + ).dontReply(); + } + + private MockMessagingSpy createFailSessionSpy(Collection<InetAddressAndPort> participants) + { + return MockMessagingService.when( + all(verb(MessagingService.Verb.REPAIR_MESSAGE), + payload((p) -> p instanceof FailSession), + to(participants::contains)) + ).dontReply(); + } + + private static RepairSessionResult createResult(CoordinatorSession coordinator) + { + return new RepairSessionResult(coordinator.sessionID, "ks", coordinator.ranges, null, false); + } +} diff --git a/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionTest.java b/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionTest.java index 5d054d3..c6980fe 100644 --- a/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionTest.java +++ b/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionTest.java @@ -374,31 +374,42 @@ public class CoordinatorSessionTest extends AbstractRepairTest coordinator.sentMessages.clear(); - // participants respond to coordinator, and repair begins once all participants have responded with success + // participants respond to coordinator, and repair begins once all participants have responded Assert.assertEquals(ConsistentSession.State.PREPARING, coordinator.getState()); coordinator.handlePrepareResponse(PARTICIPANT1, true); Assert.assertEquals(ConsistentSession.State.PREPARING, coordinator.getState()); + Assert.assertEquals(PREPARED, coordinator.getParticipantState(PARTICIPANT1)); + Assert.assertFalse(sessionResult.isDone()); // participant 2 fails to prepare for consistent repair Assert.assertFalse(coordinator.failCalled); coordinator.handlePrepareResponse(PARTICIPANT2, false); - Assert.assertEquals(ConsistentSession.State.FAILED, coordinator.getState()); - Assert.assertTrue(coordinator.failCalled); + Assert.assertEquals(ConsistentSession.State.PREPARING, coordinator.getState()); + // we should have sent failure messages to the other participants, but not yet marked them failed internally + assertMessageSent(coordinator, PARTICIPANT1, new FailSession(coordinator.sessionID)); + assertMessageSent(coordinator, PARTICIPANT2, new FailSession(coordinator.sessionID)); + assertMessageSent(coordinator, PARTICIPANT3, new FailSession(coordinator.sessionID)); + Assert.assertEquals(FAILED, coordinator.getParticipantState(PARTICIPANT2)); + Assert.assertEquals(PREPARED, coordinator.getParticipantState(PARTICIPANT1)); + Assert.assertEquals(PREPARING, coordinator.getParticipantState(PARTICIPANT3)); + Assert.assertFalse(sessionResult.isDone()); + Assert.assertFalse(coordinator.failCalled); + coordinator.sentMessages.clear(); - // additional success messages should be ignored + // last outstanding response should cause repair to complete in failed state Assert.assertFalse(coordinator.setRepairingCalled); coordinator.onSetRepairing = Assert::fail; coordinator.handlePrepareResponse(PARTICIPANT3, true); + Assert.assertTrue(coordinator.failCalled); Assert.assertFalse(coordinator.setRepairingCalled); Assert.assertFalse(repairSubmitted.get()); - // all participants should have been notified of session failure - for (InetAddressAndPort participant : PARTICIPANTS) - { - RepairMessage expected = new FailSession(coordinator.sessionID); - assertMessageSent(coordinator, participant, expected); - } + // all participants that did not fail should have been notified of session failure + RepairMessage expected = new FailSession(coordinator.sessionID); + assertMessageSent(coordinator, PARTICIPANT1, expected); + assertMessageSent(coordinator, PARTICIPANT3, expected); + Assert.assertFalse(coordinator.sentMessages.containsKey(PARTICIPANT2)); Assert.assertTrue(sessionResult.isDone()); Assert.assertTrue(hasFailures.get()); diff --git a/test/unit/org/apache/cassandra/repair/consistent/LocalSessionTest.java b/test/unit/org/apache/cassandra/repair/consistent/LocalSessionTest.java index e387c41..a6b4fe2 100644 --- a/test/unit/org/apache/cassandra/repair/consistent/LocalSessionTest.java +++ b/test/unit/org/apache/cassandra/repair/consistent/LocalSessionTest.java @@ -27,6 +27,7 @@ import java.util.Map; import java.util.UUID; import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BooleanSupplier; import java.util.function.Consumer; import com.google.common.collect.Lists; @@ -139,7 +140,8 @@ public class LocalSessionTest extends AbstractRepairTest UUID sessionID, Collection<ColumnFamilyStore> tables, RangesAtEndpoint ranges, - ExecutorService executor) + ExecutorService executor, + BooleanSupplier isCancelled) { prepareSessionCalled = true; if (prepareSessionFuture != null) @@ -148,7 +150,7 @@ public class LocalSessionTest extends AbstractRepairTest } else { - return super.prepareSession(repairManager, sessionID, tables, ranges, executor); + return super.prepareSession(repairManager, sessionID, tables, ranges, executor, isCancelled); } } @@ -334,7 +336,41 @@ public class LocalSessionTest extends AbstractRepairTest InstrumentedLocalSessions sessions = new InstrumentedLocalSessions(); sessions.handlePrepareMessage(PARTICIPANT1, new PrepareConsistentRequest(sessionID, COORDINATOR, PARTICIPANTS)); Assert.assertNull(sessions.getSession(sessionID)); - assertMessagesSent(sessions, COORDINATOR, new FailSession(sessionID)); + assertMessagesSent(sessions, COORDINATOR, new PrepareConsistentResponse(sessionID, PARTICIPANT1, false)); + } + + /** + * If the session is cancelled mid-prepare, the isCancelled boolean supplier should start returning true + */ + @Test + public void prepareCancellation() + { + UUID sessionID = registerSession(); + AtomicReference<BooleanSupplier> isCancelledRef = new AtomicReference<>(); + SettableFuture future = SettableFuture.create(); + + InstrumentedLocalSessions sessions = new InstrumentedLocalSessions() { + ListenableFuture prepareSession(KeyspaceRepairManager repairManager, UUID sessionID, Collection<ColumnFamilyStore> tables, RangesAtEndpoint ranges, ExecutorService executor, BooleanSupplier isCancelled) + { + isCancelledRef.set(isCancelled); + return future; + } + }; + sessions.start(); + + sessions.handlePrepareMessage(PARTICIPANT1, new PrepareConsistentRequest(sessionID, COORDINATOR, PARTICIPANTS)); + + BooleanSupplier isCancelled = isCancelledRef.get(); + Assert.assertNotNull(isCancelled); + Assert.assertFalse(isCancelled.getAsBoolean()); + Assert.assertTrue(sessions.sentMessages.isEmpty()); + + sessions.failSession(sessionID, false); + Assert.assertTrue(isCancelled.getAsBoolean()); + + // now that the session has failed, it send a negative response to the coordinator (even if the anti-compaction completed successfully) + future.set(new Object()); + assertMessagesSent(sessions, COORDINATOR, new PrepareConsistentResponse(sessionID, PARTICIPANT1, false)); } @Test --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@cassandra.apache.org For additional commands, e-mail: commits-h...@cassandra.apache.org