Merge branch 'cassandra-2.2' into cassandra-3.0
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/5ee6e7bc Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/5ee6e7bc Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/5ee6e7bc Branch: refs/heads/cassandra-3.0 Commit: 5ee6e7bc12a22744db462106abe1372a72b07a41 Parents: 458b36b 225232a Author: Marcus Eriksson <marc...@apache.org> Authored: Thu Jun 23 11:26:43 2016 +0200 Committer: Marcus Eriksson <marc...@apache.org> Committed: Thu Jun 23 11:26:43 2016 +0200 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../db/compaction/CompactionManager.java | 32 ++-- .../repair/RepairMessageVerbHandler.java | 31 ++-- .../cassandra/service/ActiveRepairService.java | 175 +++++++++++++++++-- .../service/ActiveRepairServiceTest.java | 125 +++++++++++-- 5 files changed, 305 insertions(+), 59 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/5ee6e7bc/CHANGES.txt ---------------------------------------------------------------------- diff --cc CHANGES.txt index ee2f6d3,b366d21..f7e854d --- a/CHANGES.txt +++ b/CHANGES.txt @@@ -6,35 -2,9 +6,36 @@@ Merged from 2.2 * Allow nodetool info to run with readonly JMX access (CASSANDRA-11755) * Validate bloom_filter_fp_chance against lowest supported value when the table is created (CASSANDRA-11920) - * RandomAccessReader: call isEOF() only when rebuffering, not for every read operation (CASSANDRA-12013) * Don't send erroneous NEW_NODE notifications on restart (CASSANDRA-11038) * StorageService shutdown hook should use a volatile variable (CASSANDRA-11984) +Merged from 2.1: ++ * Avoid marking too many sstables as repaired (CASSANDRA-11696) + * Prevent select statements with clustering key > 64k (CASSANDRA-11882) + * Fix clock skew corrupting other nodes with paxos (CASSANDRA-11991) + * Remove distinction between non-existing static columns and existing but null in LWTs (CASSANDRA-9842) + * Cache local ranges when calculating repair neighbors (CASSANDRA-11934) + * Allow LWT operation on static column with only partition keys (CASSANDRA-10532) + * Create interval tree over canonical sstables to avoid missing sstables during streaming (CASSANDRA-11886) + * cqlsh COPY FROM: shutdown parent cluster after forking, to avoid corrupting SSL connections (CASSANDRA-11749) + + +3.0.7 + * Fix legacy serialization of Thrift-generated non-compound range tombstones + when communicating with 2.x nodes (CASSANDRA-11930) + * Fix Directories instantiations where CFS.initialDirectories should be used (CASSANDRA-11849) + * Avoid referencing DatabaseDescriptor in AbstractType (CASSANDRA-11912) + * Fix sstables not being protected from removal during index build (CASSANDRA-11905) + * cqlsh: Suppress stack trace from Read/WriteFailures (CASSANDRA-11032) + * Remove unneeded code to repair index summaries that have + been improperly down-sampled (CASSANDRA-11127) + * Avoid WriteTimeoutExceptions during commit log replay due to materialized + view lock contention (CASSANDRA-11891) + * Prevent OOM failures on SSTable corruption, improve tests for corruption detection (CASSANDRA-9530) + * Use CFS.initialDirectories when clearing snapshots (CASSANDRA-11705) + * Allow compaction strategies to disable early open (CASSANDRA-11754) + * Refactor Materialized View code (CASSANDRA-11475) + * Update Java Driver (CASSANDRA-11615) +Merged from 2.2: * Persist local metadata earlier in startup sequence (CASSANDRA-11742) * Run CommitLog tests with different compression settings (CASSANDRA-9039) * cqlsh: fix tab completion for case-sensitive identifiers (CASSANDRA-11664) http://git-wip-us.apache.org/repos/asf/cassandra/blob/5ee6e7bc/src/java/org/apache/cassandra/db/compaction/CompactionManager.java ---------------------------------------------------------------------- diff --cc src/java/org/apache/cassandra/db/compaction/CompactionManager.java index bf7bd81,cf82498..99e0fd5 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@@ -1059,10 -1068,16 +1059,17 @@@ public class CompactionManager implemen try { - String snapshotName = validator.desc.sessionId.toString(); int gcBefore; + int nowInSec = FBUtilities.nowInSeconds(); + UUID parentRepairSessionId = validator.desc.parentSessionId; + String snapshotName; + boolean isGlobalSnapshotValidation = cfs.snapshotExists(parentRepairSessionId.toString()); + if (isGlobalSnapshotValidation) + snapshotName = parentRepairSessionId.toString(); + else + snapshotName = validator.desc.sessionId.toString(); boolean isSnapshotValidation = cfs.snapshotExists(snapshotName); + if (isSnapshotValidation) { // If there is a snapshot created for the session then read from there. @@@ -1083,40 -1098,53 +1090,42 @@@ StorageService.instance.forceKeyspaceFlush(cfs.keyspace.getName(), cfs.name); sstables = getSSTablesToValidate(cfs, validator); if (sstables == null) -- return; // this means the parent repair session was removed - the repair session failed on another node and we removed it ++ return; // this means the parent repair session was removed - the repair session failed on another node and we removed i if (validator.gcBefore > 0) gcBefore = validator.gcBefore; else - gcBefore = getDefaultGcBefore(cfs); + gcBefore = getDefaultGcBefore(cfs, nowInSec); } - // Create Merkle tree suitable to hold estimated partitions for given range. - // We blindly assume that partition is evenly distributed on all sstables for now. - long numPartitions = 0; - for (SSTableReader sstable : sstables) - { - numPartitions += sstable.estimatedKeysForRanges(singleton(validator.desc.range)); - } + // Create Merkle trees suitable to hold estimated partitions for the given ranges. + // We blindly assume that a partition is evenly distributed on all sstables for now. // determine tree depth from number of partitions, but cap at 20 to prevent large tree. - int depth = numPartitions > 0 ? (int) Math.min(Math.floor(Math.log(numPartitions)), 20) : 0; - MerkleTree tree = new MerkleTree(cfs.partitioner, validator.desc.range, MerkleTree.RECOMMENDED_DEPTH, (int) Math.pow(2, depth)); - + MerkleTrees tree = createMerkleTrees(sstables, validator.desc.ranges, cfs); long start = System.nanoTime(); - try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategy().getScanners(sstables, validator.desc.range)) + try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategyManager().getScanners(sstables, validator.desc.ranges); + ValidationCompactionController controller = new ValidationCompactionController(cfs, gcBefore); + CompactionIterator ci = new ValidationCompactionIterator(scanners.scanners, controller, nowInSec, metrics)) { - CompactionIterable ci = new ValidationCompactionIterable(cfs, scanners.scanners, gcBefore); - Iterator<AbstractCompactedRow> iter = ci.iterator(); - metrics.beginCompaction(ci); - try + // validate the CF as we iterate over it + validator.prepare(cfs, tree); + while (ci.hasNext()) { - // validate the CF as we iterate over it - validator.prepare(cfs, tree); - while (iter.hasNext()) + if (ci.isStopRequested()) + throw new CompactionInterruptedException(ci.getCompactionInfo()); + try (UnfilteredRowIterator partition = ci.next()) { - if (ci.isStopRequested()) - throw new CompactionInterruptedException(ci.getCompactionInfo()); - AbstractCompactedRow row = iter.next(); - validator.add(row); + validator.add(partition); } - validator.complete(); } - finally + validator.complete(); + } + finally + { - if (isSnapshotValidation) ++ if (isSnapshotValidation && !isGlobalSnapshotValidation) { + // we can only clear the snapshot if we are not doing a global snapshot validation (we then clear it once anticompaction + // is done). - if (isSnapshotValidation && !isGlobalSnapshotValidation) - { - cfs.clearSnapshot(snapshotName); - } - - metrics.finishCompaction(ci); + cfs.clearSnapshot(snapshotName); } } @@@ -1174,7 -1175,13 +1183,11 @@@ if (prs == null) return null; Set<SSTableReader> sstablesToValidate = new HashSet<>(); - + if (prs.isGlobal) + prs.markSSTablesRepairing(cfs.metadata.cfId, validator.desc.parentSessionId); - + // note that we always grab all existing sstables for this - if we were to just grab the ones that + // were marked as repairing, we would miss any ranges that were compacted away and this would cause us to overstream - try (ColumnFamilyStore.RefViewFragment sstableCandidates = cfs.selectAndReference(prs.isIncremental ? ColumnFamilyStore.UNREPAIRED_SSTABLES : ColumnFamilyStore.CANONICAL_SSTABLES)) + try (ColumnFamilyStore.RefViewFragment sstableCandidates = cfs.selectAndReference(View.select(SSTableSet.CANONICAL, (s) -> !prs.isIncremental || !s.isRepaired()))) { for (SSTableReader sstable : sstableCandidates.sstables) { http://git-wip-us.apache.org/repos/asf/cassandra/blob/5ee6e7bc/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java ---------------------------------------------------------------------- diff --cc src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java index c536b13,1701e9a..edcb4f9 --- a/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java +++ b/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java @@@ -90,27 -99,23 +90,22 @@@ public class RepairMessageVerbHandler i desc.keyspace, desc.columnFamily), message.from, id); return; } - final Collection<Range<Token>> repairingRange = desc.ranges; - Set<SSTableReader> snapshottedSSSTables = cfs.snapshot(desc.sessionId.toString(), new Predicate<SSTableReader>() + ActiveRepairService.ParentRepairSession prs = ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId); + if (prs.isGlobal) { - public boolean apply(SSTableReader sstable) - { - return sstable != null && - !sstable.metadata.isIndex() && // exclude SSTables from 2i - new Bounds<>(sstable.first.getToken(), sstable.last.getToken()).intersects(repairingRange); - } - }, true); //ephemeral snapshot, if repair fails, it will be cleaned next startup - if (ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId).isGlobal) + prs.maybeSnapshot(cfs.metadata.cfId, desc.parentSessionId); + } + else { - Set<SSTableReader> currentlyRepairing = ActiveRepairService.instance.currentlyRepairing(cfs.metadata.cfId, desc.parentSessionId); - if (!Sets.intersection(currentlyRepairing, snapshottedSSSTables).isEmpty()) - final Range<Token> repairingRange = desc.range; + cfs.snapshot(desc.sessionId.toString(), new Predicate<SSTableReader>() { - // clear snapshot that we just created - cfs.clearSnapshot(desc.sessionId.toString()); - logErrorAndSendFailureResponse("Cannot start multiple repair sessions over the same sstables", message.from, id); - return; - } - ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId).addSSTables(cfs.metadata.cfId, snapshottedSSSTables); + public boolean apply(SSTableReader sstable) + { + return sstable != null && - !(sstable.partitioner instanceof LocalPartitioner) && // exclude SSTables from 2i - new Bounds<>(sstable.first.getToken(), sstable.last.getToken()).intersects(Collections.singleton(repairingRange)); ++ !sstable.metadata.isIndex() && // exclude SSTables from 2i ++ new Bounds<>(sstable.first.getToken(), sstable.last.getToken()).intersects(desc.ranges); + } + }, true); //ephemeral snapshot, if repair fails, it will be cleaned next startup } logger.debug("Enqueuing response to snapshot request {} to {}", desc.sessionId, message.from); MessagingService.instance().sendReply(new MessageOut(MessagingService.Verb.INTERNAL_RESPONSE), id, message.from); http://git-wip-us.apache.org/repos/asf/cassandra/blob/5ee6e7bc/src/java/org/apache/cassandra/service/ActiveRepairService.java ---------------------------------------------------------------------- diff --cc src/java/org/apache/cassandra/service/ActiveRepairService.java index 6b1fd83,e111155..27c2424 --- a/src/java/org/apache/cassandra/service/ActiveRepairService.java +++ b/src/java/org/apache/cassandra/service/ActiveRepairService.java @@@ -36,9 -37,12 +37,11 @@@ import org.slf4j.Logger import org.slf4j.LoggerFactory; import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.db.ColumnFamily; import org.apache.cassandra.db.ColumnFamilyStore; --import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.db.compaction.CompactionManager; ++import org.apache.cassandra.db.lifecycle.SSTableSet; ++import org.apache.cassandra.db.lifecycle.View; + import org.apache.cassandra.dht.Bounds; -import org.apache.cassandra.dht.LocalPartitioner; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.gms.ApplicationState; @@@ -49,7 -53,7 +52,6 @@@ import org.apache.cassandra.gms.IFailur import org.apache.cassandra.gms.IEndpointStateChangeSubscriber; import org.apache.cassandra.gms.IFailureDetectionEventListener; import org.apache.cassandra.gms.VersionedValue; --import org.apache.cassandra.io.sstable.Component; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.locator.TokenMetadata; import org.apache.cassandra.net.IAsyncCallbackWithFailure; @@@ -460,8 -488,12 +487,12 @@@ public class ActiveRepairService implem public final boolean isGlobal; public final long repairedAt; public final InetAddress coordinator; + /** + * Indicates whether we have marked sstables as repairing. Can only be done once per table per ParentRepairSession + */ + private final Set<UUID> marked = new HashSet<>(); - public ParentRepairSession(InetAddress coordinator, List<ColumnFamilyStore> columnFamilyStores, Collection<Range<Token>> ranges, boolean isIncremental, boolean isGlobal, long repairedAt) + public ParentRepairSession(InetAddress coordinator, List<ColumnFamilyStore> columnFamilyStores, Collection<Range<Token>> ranges, boolean isIncremental, long repairedAt, boolean isGlobal) { this.coordinator = coordinator; for (ColumnFamilyStore cfs : columnFamilyStores) @@@ -471,15 -503,51 +502,51 @@@ } this.ranges = ranges; this.repairedAt = repairedAt; - this.isGlobal = isGlobal; this.isIncremental = isIncremental; + this.isGlobal = isGlobal; } + /** + * Mark sstables repairing - either all sstables or only the unrepaired ones depending on + * + * whether this is an incremental or full repair + * + * @param cfId the column family + * @param parentSessionId the parent repair session id, used to make sure we don't start multiple repairs over the same sstables + */ + public synchronized void markSSTablesRepairing(UUID cfId, UUID parentSessionId) + { + if (!marked.contains(cfId)) + { - List<SSTableReader> sstables = columnFamilyStores.get(cfId).select(isIncremental ? ColumnFamilyStore.UNREPAIRED_SSTABLES : ColumnFamilyStore.CANONICAL_SSTABLES).sstables; ++ List<SSTableReader> sstables = columnFamilyStores.get(cfId).select(View.select(SSTableSet.CANONICAL, (s) -> !isIncremental || !s.isRepaired())).sstables; + Set<SSTableReader> currentlyRepairing = ActiveRepairService.instance.currentlyRepairing(cfId, parentSessionId); + if (!Sets.intersection(currentlyRepairing, Sets.newHashSet(sstables)).isEmpty()) + { + logger.error("Cannot start multiple repair sessions over the same sstables"); + throw new RuntimeException("Cannot start multiple repair sessions over the same sstables"); + } + addSSTables(cfId, sstables); + marked.add(cfId); + } + } + + /** + * Get the still active sstables we should run anticompaction on + * + * note that validation and streaming do not call this method - they have to work on the actual active sstables on the node, we only call this + * to know which sstables are still there that were there when we started the repair + * + * @param cfId + * @param parentSessionId for checking if there exists a snapshot for this repair + * @return + */ @SuppressWarnings("resource") - public synchronized Refs<SSTableReader> getActiveRepairedSSTableRefs(UUID cfId) + public synchronized Refs<SSTableReader> getActiveRepairedSSTableRefsForAntiCompaction(UUID cfId, UUID parentSessionId) { + assert marked.contains(cfId); + boolean isSnapshotRepair = columnFamilyStores.get(cfId).snapshotExists(parentSessionId.toString()); ImmutableMap.Builder<SSTableReader, Ref<SSTableReader>> references = ImmutableMap.builder(); - for (SSTableReader sstable : getActiveSSTables(cfId)) + for (SSTableReader sstable : isSnapshotRepair ? getSSTablesForSnapshotRepair(cfId, parentSessionId) : getActiveSSTables(cfId)) { Ref<SSTableReader> ref = sstable.tryRef(); if (ref == null) @@@ -490,12 -558,97 +557,97 @@@ return new Refs<>(references.build()); } + /** + * If we are running a snapshot repair we need to find the 'real' sstables when we start anticompaction + * + * We use the generation of the sstables as identifiers instead of the file name to avoid having to parse out the + * actual filename. + * + * @param cfId + * @param parentSessionId + * @return + */ + private Set<SSTableReader> getSSTablesForSnapshotRepair(UUID cfId, UUID parentSessionId) + { + Set<SSTableReader> activeSSTables = new HashSet<>(); + ColumnFamilyStore cfs = columnFamilyStores.get(cfId); + + Set<Integer> snapshotGenerations = new HashSet<>(); + try (Refs<SSTableReader> snapshottedSSTables = cfs.getSnapshotSSTableReader(parentSessionId.toString())) + { + for (SSTableReader sstable : snapshottedSSTables) + { + snapshotGenerations.add(sstable.descriptor.generation); + } + } + catch (IOException e) + { + throw new RuntimeException(e); + } - for (SSTableReader sstable : cfs.select(ColumnFamilyStore.CANONICAL_SSTABLES).sstables) ++ for (SSTableReader sstable : cfs.getSSTables(SSTableSet.CANONICAL)) + if (snapshotGenerations.contains(sstable.descriptor.generation)) + activeSSTables.add(sstable); + return activeSSTables; + } + + public synchronized void maybeSnapshot(UUID cfId, UUID parentSessionId) + { + String snapshotName = parentSessionId.toString(); + if (!columnFamilyStores.get(cfId).snapshotExists(snapshotName)) + { + Set<SSTableReader> snapshottedSSTables = columnFamilyStores.get(cfId).snapshot(snapshotName, new Predicate<SSTableReader>() + { + public boolean apply(SSTableReader sstable) + { + return sstable != null && + (!isIncremental || !sstable.isRepaired()) && - !(sstable.partitioner instanceof LocalPartitioner) && // exclude SSTables from 2i ++ !(sstable.metadata.isIndex()) && // exclude SSTables from 2i + new Bounds<>(sstable.first.getToken(), sstable.last.getToken()).intersects(ranges); + } + }, true); + + if (isAlreadyRepairing(cfId, parentSessionId, snapshottedSSTables)) + { + columnFamilyStores.get(cfId).clearSnapshot(parentSessionId.toString()); + logger.error("Cannot start multiple repair sessions over the same sstables"); + throw new RuntimeException("Cannot start multiple repair sessions over the same sstables"); + } + addSSTables(cfId, snapshottedSSTables); + marked.add(cfId); + } + } + + + /** + * Compares other repairing sstables *generation* to the ones we just snapshotted + * + * we compare generations since the sstables have different paths due to snapshot names + * + * @param cfId id of the column family store + * @param parentSessionId parent repair session + * @param sstables the newly snapshotted sstables + * @return + */ + private boolean isAlreadyRepairing(UUID cfId, UUID parentSessionId, Collection<SSTableReader> sstables) + { + Set<SSTableReader> currentlyRepairing = ActiveRepairService.instance.currentlyRepairing(cfId, parentSessionId); + Set<Integer> currentlyRepairingGenerations = new HashSet<>(); + Set<Integer> newRepairingGenerations = new HashSet<>(); + for (SSTableReader sstable : currentlyRepairing) + currentlyRepairingGenerations.add(sstable.descriptor.generation); + for (SSTableReader sstable : sstables) + newRepairingGenerations.add(sstable.descriptor.generation); + + return !Sets.intersection(currentlyRepairingGenerations, newRepairingGenerations).isEmpty(); + } + private Set<SSTableReader> getActiveSSTables(UUID cfId) { Set<String> repairedSSTables = sstableMap.get(cfId); Set<SSTableReader> activeSSTables = new HashSet<>(); Set<String> activeSSTableNames = new HashSet<>(); - for (SSTableReader sstable : columnFamilyStores.get(cfId).getSSTables()) + ColumnFamilyStore cfs = columnFamilyStores.get(cfId); - for (SSTableReader sstable : cfs.select(ColumnFamilyStore.CANONICAL_SSTABLES).sstables) ++ for (SSTableReader sstable : cfs.getSSTables(SSTableSet.CANONICAL)) { if (repairedSSTables.contains(sstable.getFilename())) { http://git-wip-us.apache.org/repos/asf/cassandra/blob/5ee6e7bc/test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java ---------------------------------------------------------------------- diff --cc test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java index da067fd,03a25c6..adcd684 --- a/test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java +++ b/test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java @@@ -20,18 -20,23 +20,21 @@@ package org.apache.cassandra.service import java.net.InetAddress; import java.util.*; + import java.util.concurrent.ExecutionException; -import com.google.common.base.Predicate; import com.google.common.collect.Sets; import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; import org.apache.cassandra.SchemaLoader; -import org.apache.cassandra.Util; import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.config.KSMetaData; import org.apache.cassandra.db.ColumnFamilyStore; -import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.Keyspace; -import org.apache.cassandra.db.Mutation; +import org.apache.cassandra.db.RowUpdateBuilder; import org.apache.cassandra.db.compaction.OperationType; ++import org.apache.cassandra.db.lifecycle.SSTableSet; ++import org.apache.cassandra.db.lifecycle.View; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.exceptions.ConfigurationException; @@@ -225,17 -233,15 +229,15 @@@ public class ActiveRepairServiceTes public void testGetActiveRepairedSSTableRefs() { ColumnFamilyStore store = prepareColumnFamilyStore(); - Set<SSTableReader> original = store.getUnrepairedSSTables(); + Set<SSTableReader> original = store.getLiveSSTables(); UUID prsId = UUID.randomUUID(); - ActiveRepairService.instance.registerParentRepairSession(prsId, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), null, true, false); + ActiveRepairService.instance.registerParentRepairSession(prsId, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), null, true, 0, false); ActiveRepairService.ParentRepairSession prs = ActiveRepairService.instance.getParentRepairSession(prsId); - - //add all sstables to parent repair session - prs.addSSTables(store.metadata.cfId, original); + prs.markSSTablesRepairing(store.metadata.cfId, prsId); //retrieve all sstable references from parent repair sessions - Refs<SSTableReader> refs = prs.getActiveRepairedSSTableRefs(store.metadata.cfId); + Refs<SSTableReader> refs = prs.getActiveRepairedSSTableRefsForAntiCompaction(store.metadata.cfId, prsId); Set<SSTableReader> retrieved = Sets.newHashSet(refs.iterator()); assertEquals(original, retrieved); refs.release(); @@@ -261,20 -267,117 +263,117 @@@ refs.release(); } + @Test + public void testAddingMoreSSTables() + { + ColumnFamilyStore store = prepareColumnFamilyStore(); - Set<SSTableReader> original = store.getUnrepairedSSTables(); ++ Set<SSTableReader> original = Sets.newHashSet(store.select(View.select(SSTableSet.CANONICAL, (s) -> !s.isRepaired())).sstables); + UUID prsId = UUID.randomUUID(); - ActiveRepairService.instance.registerParentRepairSession(prsId, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), null, true, true); ++ ActiveRepairService.instance.registerParentRepairSession(prsId, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), null, true, System.currentTimeMillis(), true); + ActiveRepairService.ParentRepairSession prs = ActiveRepairService.instance.getParentRepairSession(prsId); + prs.markSSTablesRepairing(store.metadata.cfId, prsId); + try (Refs<SSTableReader> refs = prs.getActiveRepairedSSTableRefsForAntiCompaction(store.metadata.cfId, prsId)) + { + Set<SSTableReader> retrieved = Sets.newHashSet(refs.iterator()); + assertEquals(original, retrieved); + } + createSSTables(store, 2); + boolean exception = false; + try + { + UUID newPrsId = UUID.randomUUID(); - ActiveRepairService.instance.registerParentRepairSession(newPrsId, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), null, true, true); ++ ActiveRepairService.instance.registerParentRepairSession(newPrsId, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), null, true, System.currentTimeMillis(), true); + ActiveRepairService.instance.getParentRepairSession(newPrsId).markSSTablesRepairing(store.metadata.cfId, newPrsId); + } + catch (Throwable t) + { + exception = true; + } + assertTrue(exception); + + try (Refs<SSTableReader> refs = prs.getActiveRepairedSSTableRefsForAntiCompaction(store.metadata.cfId, prsId)) + { + Set<SSTableReader> retrieved = Sets.newHashSet(refs.iterator()); + assertEquals(original, retrieved); + } + } + + @Test + public void testSnapshotAddSSTables() throws ExecutionException, InterruptedException + { + ColumnFamilyStore store = prepareColumnFamilyStore(); + UUID prsId = UUID.randomUUID(); - Set<SSTableReader> original = store.getUnrepairedSSTables(); - ActiveRepairService.instance.registerParentRepairSession(prsId, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), Collections.singleton(new Range<>(store.partitioner.getMinimumToken(), store.partitioner.getMinimumToken())), true, true); ++ Set<SSTableReader> original = Sets.newHashSet(store.select(View.select(SSTableSet.CANONICAL, (s) -> !s.isRepaired())).sstables); ++ ActiveRepairService.instance.registerParentRepairSession(prsId, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), Collections.singleton(new Range<>(store.getPartitioner().getMinimumToken(), store.getPartitioner().getMinimumToken())), true, System.currentTimeMillis(), true); + ActiveRepairService.instance.getParentRepairSession(prsId).maybeSnapshot(store.metadata.cfId, prsId); + + UUID prsId2 = UUID.randomUUID(); - ActiveRepairService.instance.registerParentRepairSession(prsId2, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), Collections.singleton(new Range<>(store.partitioner.getMinimumToken(), store.partitioner.getMinimumToken())), true, true); ++ ActiveRepairService.instance.registerParentRepairSession(prsId2, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), Collections.singleton(new Range<>(store.getPartitioner().getMinimumToken(), store.getPartitioner().getMinimumToken())), true, System.currentTimeMillis(), true); + createSSTables(store, 2); + ActiveRepairService.instance.getParentRepairSession(prsId).maybeSnapshot(store.metadata.cfId, prsId); + try (Refs<SSTableReader> refs = ActiveRepairService.instance.getParentRepairSession(prsId).getActiveRepairedSSTableRefsForAntiCompaction(store.metadata.cfId, prsId)) + { + assertEquals(original, Sets.newHashSet(refs.iterator())); + } + store.forceMajorCompaction(); + // after a major compaction the original sstables will be gone and we will have no sstables to anticompact: + try (Refs<SSTableReader> refs = ActiveRepairService.instance.getParentRepairSession(prsId).getActiveRepairedSSTableRefsForAntiCompaction(store.metadata.cfId, prsId)) + { + assertEquals(0, refs.size()); + } + } + + @Test + public void testSnapshotMultipleRepairs() + { + ColumnFamilyStore store = prepareColumnFamilyStore(); - Set<SSTableReader> original = store.getUnrepairedSSTables(); ++ Set<SSTableReader> original = Sets.newHashSet(store.select(View.select(SSTableSet.CANONICAL, (s) -> !s.isRepaired())).sstables); + UUID prsId = UUID.randomUUID(); - ActiveRepairService.instance.registerParentRepairSession(prsId, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), Collections.singleton(new Range<>(store.partitioner.getMinimumToken(), store.partitioner.getMinimumToken())), true, true); ++ ActiveRepairService.instance.registerParentRepairSession(prsId, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), Collections.singleton(new Range<>(store.getPartitioner().getMinimumToken(), store.getPartitioner().getMinimumToken())), true, System.currentTimeMillis(), true); + ActiveRepairService.instance.getParentRepairSession(prsId).maybeSnapshot(store.metadata.cfId, prsId); + + UUID prsId2 = UUID.randomUUID(); - ActiveRepairService.instance.registerParentRepairSession(prsId2, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), Collections.singleton(new Range<>(store.partitioner.getMinimumToken(), store.partitioner.getMinimumToken())), true, true); ++ ActiveRepairService.instance.registerParentRepairSession(prsId2, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), Collections.singleton(new Range<>(store.getPartitioner().getMinimumToken(), store.getPartitioner().getMinimumToken())), true, System.currentTimeMillis(), true); + boolean exception = false; + try + { + ActiveRepairService.instance.getParentRepairSession(prsId2).maybeSnapshot(store.metadata.cfId, prsId2); + } + catch (Throwable t) + { + exception = true; + } + assertTrue(exception); + try (Refs<SSTableReader> refs = ActiveRepairService.instance.getParentRepairSession(prsId).getActiveRepairedSSTableRefsForAntiCompaction(store.metadata.cfId, prsId)) + { + assertEquals(original, Sets.newHashSet(refs.iterator())); + } + } + private ColumnFamilyStore prepareColumnFamilyStore() { Keyspace keyspace = Keyspace.open(KEYSPACE5); ColumnFamilyStore store = keyspace.getColumnFamilyStore(CF_STANDARD1); + store.truncateBlocking(); store.disableAutoCompaction(); - for (int i = 0; i < 10; i++) + createSSTables(store, 10); + return store; + } + + private void createSSTables(ColumnFamilyStore cfs, int count) + { + long timestamp = System.currentTimeMillis(); + for (int i = 0; i < count; i++) { - new RowUpdateBuilder(store.metadata, System.currentTimeMillis(), Integer.toString(i)) - .clustering("c") - .add("val", "val") - .build() - .applyUnsafe(); - DecoratedKey key = Util.dk(Integer.toString(i)); - Mutation rm = new Mutation(KEYSPACE5, key.getKey()); + for (int j = 0; j < 10; j++) - rm.add("Standard1", Util.cellname(Integer.toString(j)), - ByteBufferUtil.EMPTY_BYTE_BUFFER, - timestamp, - 0); - rm.apply(); ++ { ++ new RowUpdateBuilder(cfs.metadata, timestamp, Integer.toString(j)) ++ .clustering("c") ++ .add("val", "val") ++ .build() ++ .applyUnsafe(); ++ } + cfs.forceBlockingFlush(); } - store.forceBlockingFlush(); - return store; } }
