diff --git a/CHANGES.txt b/CHANGES.txt index 44cc5dcce5..d268b7cac7 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -3,6 +3,7 @@ * Allow nodetool garbagecollect to take a user defined list of SSTables (CASSANDRA-16767) * Add a guardrail for misprepared statements (CASSANDRA-21139) Merged from 6.0: + * Safely regain ranges and delete retired command stores (CASSANDRA-21212) * Reduce memory allocations in miscellaneous places along read path (CASSANDRA-21360) * Avoid ByteBuffer wrapping in cql3.selection.Selector.InputRow to reduce memory allocation rate (CASSANDRA-21362) * Reduce cost to calculate BTreeRow.minDeletionTime (CASSANDRA-21414) diff --git a/modules/accord b/modules/accord index 3d6392f4ec..0a10cd0567 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 3d6392f4eca41d3036a704f1513020c24689f1bb +Subproject commit 0a10cd056794c05588114f45ce86d49d6d6538db diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java index d0ad79cea3..72f7ba4363 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java @@ -1008,9 +1008,9 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte if (read.kind() == Repeat && !hasWritten) { Invariants.require(lastImage != null); - write = new TopologyImage(read.epoch(), Image, lastImage.getUpdate()); + write = new TopologyImage(read.epoch(), Image, lastImage.update()); } - else if (hasWritten && read.kind() == Repeat && lastImage.getUpdate().isEquivalent(read.getUpdate())) + else if (hasWritten && read.kind() == Repeat && lastImage.update().isEquivalent(read.update())) { write = read.asRepeat(); } diff --git a/src/java/org/apache/cassandra/service/accord/AccordDurableOnFlush.java b/src/java/org/apache/cassandra/service/accord/AccordDurableOnFlush.java index a7dfd2a15f..6905b10bf9 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordDurableOnFlush.java +++ b/src/java/org/apache/cassandra/service/accord/AccordDurableOnFlush.java @@ -127,7 +127,9 @@ class AccordDurableOnFlush implements BiConsumer for (Map.Entry e : notify.entrySet()) { ReportDurable durable = e.getValue(); - notifyInOrder(memtableId, metadata, commandStores.forId(e.getKey()), durable); + CommandStore commandStore = commandStores.forId(e.getKey()); + if (commandStore != null) + notifyInOrder(memtableId, metadata, commandStore, durable); } } diff --git a/src/java/org/apache/cassandra/service/accord/JournalKey.java b/src/java/org/apache/cassandra/service/accord/JournalKey.java index b4731e2496..f907d51418 100644 --- a/src/java/org/apache/cassandra/service/accord/JournalKey.java +++ b/src/java/org/apache/cassandra/service/accord/JournalKey.java @@ -40,6 +40,7 @@ import static org.apache.cassandra.db.TypeSizes.LONG_SIZE; import static org.apache.cassandra.service.accord.journal.MergeSerializers.BootstrapBeganAtSerializer; import static org.apache.cassandra.service.accord.journal.MergeSerializers.CommandChangeSerializer; import static org.apache.cassandra.service.accord.journal.MergeSerializers.DurableBeforeSerializer; +import static org.apache.cassandra.service.accord.journal.MergeSerializers.PermanentlyUnsafeToReadSerializer; import static org.apache.cassandra.service.accord.journal.MergeSerializers.RangesForEpochSerializer; import static org.apache.cassandra.service.accord.journal.MergeSerializers.RedundantBeforeSerializer; import static org.apache.cassandra.service.accord.journal.MergeSerializers.SafeToReadSerializer; @@ -273,7 +274,8 @@ public final class JournalKey SAFE_TO_READ (3, new SafeToReadSerializer(), false), BOOTSTRAP_BEGAN_AT (4, new BootstrapBeganAtSerializer(), false), RANGES_FOR_EPOCH (5, new RangesForEpochSerializer(), false), - TOPOLOGY_UPDATE (6, new TopologySerializer(), true), + TOPOLOGY_UPDATE (6, TopologySerializer.INSTANCE, true), + PERMANENTLY_UNSAFE_TO_READ (7, new PermanentlyUnsafeToReadSerializer(), false) ; public final int id; diff --git a/src/java/org/apache/cassandra/service/accord/journal/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/journal/AccordJournal.java index 425aa0f4f2..ee728a8778 100644 --- a/src/java/org/apache/cassandra/service/accord/journal/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/journal/AccordJournal.java @@ -288,12 +288,13 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier logger.error("Encountered TopologyImage Repeat record for epoch {}, but no prior image record was found", ref.key().id.epoch()); return null; } - prev = reader.read().asImage(Invariants.nonNull(prev.getUpdate())); + prev = reader.read().asImage(Invariants.nonNull(prev.update())); } else prev = reader.read(); - return new accord.api.Journal.TopologyUpdate(prev.getUpdate().commandStores, - prev.getUpdate().global); + return new accord.api.Journal.TopologyUpdate(prev.update().commandStores, + prev.update().global, + prev.update().previouslyOwned); } @Override @@ -358,6 +359,13 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier return accumulator.get(); } + @Override + public Ranges loadPermanentlyUnsafeToRead(int commandStoreId) + { + KeepFirst accumulator = readLast(new JournalKey(TxnId.NONE, JournalKey.Type.PERMANENTLY_UNSAFE_TO_READ, commandStoreId)); + return accumulator.get(); + } + @Override public PersistentField.Persister durableBeforePersister() { @@ -376,6 +384,8 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.SAFE_TO_READ, commandStoreId), fieldUpdates.newSafeToRead); if (fieldUpdates.newRangesForEpoch != null) pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.RANGES_FOR_EPOCH, commandStoreId), fieldUpdates.newRangesForEpoch); + if (fieldUpdates.newPermanentlyUnsafeToRead != null) + pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.PERMANENTLY_UNSAFE_TO_READ, commandStoreId), fieldUpdates.newPermanentlyUnsafeToRead); if (onFlush == null) return; diff --git a/src/java/org/apache/cassandra/service/accord/journal/MergeSerializers.java b/src/java/org/apache/cassandra/service/accord/journal/MergeSerializers.java index 74b00f740c..92a5ee044a 100644 --- a/src/java/org/apache/cassandra/service/accord/journal/MergeSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/journal/MergeSerializers.java @@ -37,6 +37,7 @@ import org.apache.cassandra.service.accord.JournalKey; import org.apache.cassandra.service.accord.journal.Merger.KeepFirst; import org.apache.cassandra.service.accord.journal.Merger.SimpleMerger; import org.apache.cassandra.service.accord.serializers.CommandStoreSerializers; +import org.apache.cassandra.service.accord.serializers.KeySerializers; import org.apache.cassandra.service.accord.serializers.Version; import static accord.local.CommandStores.RangesForEpoch; @@ -250,7 +251,6 @@ public class MergeSerializers SimpleMerger, SimpleMerger> { - public static final RangesForEpochSerializer instance = new RangesForEpochSerializer(); public KeepFirst mergerFor() { return new KeepFirst<>(null); @@ -275,11 +275,40 @@ public class MergeSerializers } } + public static class PermanentlyUnsafeToReadSerializer + implements MergeSerializer, + SimpleMerger> + { + public KeepFirst mergerFor() + { + return new KeepFirst<>(Ranges.EMPTY); + } + + @Override + public void serialize(JournalKey key, Ranges from, DataOutputPlus out, Version userVersion) throws IOException + { + KeySerializers.ranges.serialize(from, out); + } + + @Override + public void reserialize(JournalKey key, SimpleMerger from, DataOutputPlus out, Version userVersion) throws IOException + { + serialize(key, from.get(), out, userVersion); + } + + @Override + public void deserialize(JournalKey key, SimpleMerger into, DataInputPlus in, Version userVersion) throws IOException + { + into.update(KeySerializers.ranges.deserialize(in)); + } + } + public static class TopologySerializer implements MergeSerializer { public static final TopologySerializer INSTANCE = new TopologySerializer(); - public TopologySerializer() {} + private TopologySerializer() {} @Override public TopologyMerger mergerFor() @@ -309,6 +338,7 @@ public class MergeSerializers public static class TopologyMerger implements Merger { TopologyRecord.TopologyImage read, write; + boolean hasRead; public TopologyMerger() { @@ -317,6 +347,7 @@ public class MergeSerializers @Override public void reset(JournalKey key) { + hasRead = false; read = write = null; } @@ -327,11 +358,15 @@ public class MergeSerializers public void read(TopologyRecord update) { + if (hasRead) + return; + if (Objects.requireNonNull(update.kind()) == TopologyRecord.Kind.New) - read = new TopologyRecord.TopologyImage(update.epoch(), TopologyRecord.Kind.Image, update.getUpdate()); + read = new TopologyRecord.TopologyImage(update.epoch(), TopologyRecord.Kind.Image, update.update()); else read = (TopologyRecord.TopologyImage) update; write = read; + hasRead = true; } public void write(TopologyRecord.TopologyImage image) diff --git a/src/java/org/apache/cassandra/service/accord/journal/TopologyRecord.java b/src/java/org/apache/cassandra/service/accord/journal/TopologyRecord.java index c59a184170..98ede3c35e 100644 --- a/src/java/org/apache/cassandra/service/accord/journal/TopologyRecord.java +++ b/src/java/org/apache/cassandra/service/accord/journal/TopologyRecord.java @@ -26,6 +26,7 @@ import org.agrona.collections.Int2ObjectHashMap; import accord.api.Journal; import accord.local.CommandStores; +import accord.local.CommandStores.PreviouslyOwned; import accord.local.Node; import accord.primitives.Ranges; import accord.primitives.TxnId; @@ -50,7 +51,7 @@ public interface TopologyRecord long epoch(); TopologyRecord asRepeat(); - Journal.TopologyUpdate getUpdate(); + Journal.TopologyUpdate update(); static TopologyRecord newTopology(Journal.TopologyUpdate update) { return new NewTopology(update); @@ -121,7 +122,7 @@ public interface TopologyRecord } @Override - public Journal.TopologyUpdate getUpdate() + public Journal.TopologyUpdate update() { return update; } @@ -182,7 +183,7 @@ public interface TopologyRecord } @Override - public Journal.TopologyUpdate getUpdate() + public Journal.TopologyUpdate update() { return update; } @@ -225,12 +226,20 @@ public interface TopologyRecord class TopologyUpdateSerializer implements UnversionedSerializer { + private static final int TOP_BIT = 0x40000000; public static final TopologyUpdateSerializer instance = new TopologyUpdateSerializer(); @Override public void serialize(Journal.TopologyUpdate from, DataOutputPlus out) throws IOException { - out.writeUnsignedVInt32(from.commandStores.size()); + out.writeUnsignedVInt32(from.commandStores.size() | TOP_BIT); + out.writeUnsignedVInt32(0); + out.writeUnsignedVInt32(from.previouslyOwned.size()); + for (int i = 0 ; i < from.previouslyOwned.size() ; ++i) + { + out.writeUnsignedVInt(from.previouslyOwned.epochs(i)); + KeySerializers.ranges.serialize(from.previouslyOwned.ranges(i), out); + } for (Map.Entry e : from.commandStores.entrySet()) { out.writeUnsignedVInt32(e.getKey()); @@ -243,6 +252,23 @@ public interface TopologyRecord public Journal.TopologyUpdate deserialize(DataInputPlus in) throws IOException { int commandStoresSize = in.readUnsignedVInt32(); + int flags = 0; + PreviouslyOwned previouslyOwned = PreviouslyOwned.EMPTY; + if ((commandStoresSize & TOP_BIT) != 0) + { + commandStoresSize ^= TOP_BIT; + // future proofing + flags = in.readUnsignedVInt32(); + int previouslyOwnedSize = in.readUnsignedVInt32(); + long[] epochs = new long[previouslyOwnedSize]; + Ranges[] ranges = new Ranges[previouslyOwnedSize]; + for (int i = 0 ; i < previouslyOwnedSize ; ++i) + { + epochs[i] = in.readUnsignedVInt(); + ranges[i] = KeySerializers.ranges.deserialize(in); + } + previouslyOwned = new PreviouslyOwned(epochs.length == 0 ? 0 : epochs[0], epochs, ranges); + } Int2ObjectHashMap commandStores = new Int2ObjectHashMap<>(); for (int j = 0; j < commandStoresSize; j++) { @@ -251,13 +277,20 @@ public interface TopologyRecord commandStores.put(commandStoreId, rfe); } Topology global = TopologySerializers.compactTopology.deserialize(in); - return new Journal.TopologyUpdate(commandStores, global); + return new Journal.TopologyUpdate(commandStores, global, previouslyOwned); } @Override public long serializedSize(Journal.TopologyUpdate from) { - long size = TypeSizes.sizeofUnsignedVInt(from.commandStores.size()); + long size = TypeSizes.sizeofUnsignedVInt(from.commandStores.size() | TOP_BIT); + size += TypeSizes.sizeofUnsignedVInt(0); + size += TypeSizes.sizeofUnsignedVInt(from.previouslyOwned.size()); + for (int i = 0 ; i < from.previouslyOwned.size() ; ++i) + { + size += TypeSizes.sizeofUnsignedVInt(from.previouslyOwned.epochs(i)); + size += KeySerializers.ranges.serializedSize(from.previouslyOwned.ranges(i)); + } for (Map.Entry e : from.commandStores.entrySet()) { size += TypeSizes.sizeofUnsignedVInt(e.getKey()); diff --git a/src/java/org/apache/cassandra/tcm/sequences/Move.java b/src/java/org/apache/cassandra/tcm/sequences/Move.java index a7f316e38f..b4c55ba348 100644 --- a/src/java/org/apache/cassandra/tcm/sequences/Move.java +++ b/src/java/org/apache/cassandra/tcm/sequences/Move.java @@ -27,12 +27,20 @@ import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.concurrent.atomic.AtomicBoolean; +import javax.annotation.Nullable; + import com.google.common.annotations.VisibleForTesting; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import accord.api.TopologyListener; +import accord.primitives.Ranges; +import accord.topology.ActiveEpoch; import accord.topology.EpochReady; +import accord.topology.Topology; +import accord.topology.TopologyManager; +import accord.topology.TopologyManager.RegainingEpochRange; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.SystemKeyspace; @@ -54,6 +62,7 @@ import org.apache.cassandra.schema.ReplicationParams; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.service.accord.AccordService; +import org.apache.cassandra.service.accord.topology.AccordTopology; import org.apache.cassandra.streaming.StreamOperation; import org.apache.cassandra.streaming.StreamPlan; import org.apache.cassandra.streaming.StreamResultFuture; @@ -75,10 +84,13 @@ import org.apache.cassandra.tcm.serialization.Version; import org.apache.cassandra.tcm.transformations.PrepareMove; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.JVMStabilityInspector; +import org.apache.cassandra.utils.concurrent.Condition; import org.apache.cassandra.utils.concurrent.Future; import org.apache.cassandra.utils.concurrent.FutureCombiner; import org.apache.cassandra.utils.vint.VIntCoding; +import static accord.primitives.AbstractRanges.UnionMode.MERGE_ADJACENT; +import static accord.primitives.Routables.Slice.Minimal; import static com.google.common.collect.ImmutableList.of; import static org.apache.cassandra.tcm.MultiStepOperation.Kind.MOVE; import static org.apache.cassandra.tcm.Transformation.Kind.FINISH_MOVE; @@ -86,6 +98,7 @@ import static org.apache.cassandra.tcm.Transformation.Kind.MID_MOVE; import static org.apache.cassandra.tcm.Transformation.Kind.START_MOVE; import static org.apache.cassandra.tcm.sequences.SequenceState.continuable; import static org.apache.cassandra.tcm.sequences.SequenceState.error; +import static org.apache.cassandra.utils.concurrent.Condition.newOneTimeCondition; public class Move extends MultiStepOperation { @@ -196,6 +209,44 @@ public class Move extends MultiStepOperation return applyMultipleTransformations(metadata, next, of(startMove, midMove, finishMove)); } + static class WaitForEpochAndRangeRetirement implements TopologyListener + { + final Condition condition = newOneTimeCondition(); + final long waitingForEpoch; + final Ranges waitingForRanges; + Ranges retiredRanges; + + public WaitForEpochAndRangeRetirement(long waitingForEpoch, Ranges waitingForRanges) + { + this.waitingForEpoch = waitingForEpoch; + this.waitingForRanges = waitingForRanges; + this.retiredRanges = Ranges.EMPTY; + } + + synchronized void updateRetiredRanges(Ranges ranges) + { + ranges = ranges.slice(waitingForRanges, Minimal).without(retiredRanges); + if (!ranges.isEmpty()) + { + retiredRanges = retiredRanges.union(MERGE_ADJACENT, ranges); + if (retiredRanges.containsAll(waitingForRanges)) + condition.signal(); + } + } + + @Override + public synchronized void onEpochRetired(Ranges ranges, long epoch, @Nullable Topology topology) + { + if (epoch >= waitingForEpoch) + updateRetiredRanges(ranges); + } + + public void waitForRetirement() + { + condition.awaitThrowUncheckedOnInterrupt(); + } + } + @Override public Set affectedPeers(Directory directory) { @@ -212,9 +263,38 @@ public class Move extends MultiStepOperation switch (next) { case START_MOVE: + WaitForEpochAndRangeRetirement wait = null; + try { ClusterMetadata metadata = ClusterMetadata.current(); + if (AccordService.isStarted() && metadata.schema.hasAccordKeyspaces()) + { + TopologyManager topologyManager = AccordService.instance().topology(); + AccordService.toFuture(topologyManager.await(metadata.epoch.getEpoch(), null)) + .awaitThrowUncheckedOnInterrupt().rethrowIfFailed(); + Topology current = topologyManager.active().globalForEpoch(metadata.epoch.getEpoch()); + RegainingEpochRange regaining = topologyManager.computeRegaining(current, AccordTopology.createAccordTopology(applyTo(metadata).success().metadata)); + + if (regaining != null) + { + wait = new WaitForEpochAndRangeRetirement(regaining.epoch(), regaining.ranges()); + topologyManager.addListener(wait); + ActiveEpoch e = topologyManager.active().ifExists(regaining.epoch()); + + // We have already checked that our activeEpochs is caught up to metadata.epoch.getEpoch() + // which is greater than regaining.epoch() so if e == null then the only case we can fall + // in is that regaining.epoch() is already retired + if (e != null) + { + wait.updateRetiredRanges(e.retired()); + logger.info("Waiting for previous ownership of ranges {} to retire before regaining", regaining.ranges()); + wait.waitForRetirement(); + logger.info("Previously owned ranges {} now retired", regaining.ranges()); + } + } + } + logger.info("Moving {} from {} to {}.", metadata.directory.endpoint(startMove.nodeId()), metadata.tokenMap.tokens(startMove.nodeId()), @@ -226,6 +306,11 @@ public class Move extends MultiStepOperation JVMStabilityInspector.inspectThrowable(t); return continuable(); } + finally + { + if (wait != null) + AccordService.instance().topology().removeListener(wait); + } break; case MID_MOVE: try diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordDeleteCommandStoreTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordDeleteCommandStoreTest.java new file mode 100644 index 0000000000..21713e5a56 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordDeleteCommandStoreTest.java @@ -0,0 +1,81 @@ +/* + * 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.distributed.test.accord; + +import java.io.IOException; +import java.util.Arrays; +import java.util.List; + +import org.junit.BeforeClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.Util; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.service.accord.AccordService; + +import static com.google.common.collect.Iterables.getOnlyElement; + +public class AccordDeleteCommandStoreTest extends AccordTestBase +{ + private static final Logger logger = LoggerFactory.getLogger(AccordDeleteCommandStoreTest.class); + + @Override + protected Logger logger() + { + return logger; + } + + @BeforeClass + public static void setupClass() throws IOException + { + AccordTestBase.setupCluster(builder -> builder + .withoutVNodes() + .withConfig(config -> + config + .set("accord.shard_durability_cycle", "20s") + .set("accord.topology_sync_propagator_enabled_pre_start", true) + .with(Feature.NETWORK, Feature.GOSSIP)), 4); + } + + @Test + public void deleteCommandStoresTest() throws Throwable + { + List ddls = Arrays.asList("DROP KEYSPACE IF EXISTS " + KEYSPACE + ';', + "CREATE KEYSPACE " + KEYSPACE + " WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor': 2}", + "CREATE TABLE " + qualifiedAccordTableName + " (k int PRIMARY KEY, v int) WITH transactional_mode='full'"); + test(ddls, cluster -> { + String newToken = cluster.get(1).callOnInstance(() -> getOnlyElement(StorageService.instance.getTokens())); + + int numberOfCommandStores = cluster.get(2).callOnInstance(() -> { + Util.spinUntilTrue(() -> AccordService.instance().node().commandStores().all().length > 0); + return AccordService.instance().node().commandStores().all().length; + }); + + cluster.get(2).runOnInstance(() -> { + StorageService.instance.move(Long.toString(Long.parseLong(newToken) + 1)); + Util.spinUntilTrue(() -> AccordService.instance().node().commandStores().all().length < numberOfCommandStores, + 60); + }); + }); + } +} + diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordRegainRangesTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordRegainRangesTest.java new file mode 100644 index 0000000000..990c875024 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordRegainRangesTest.java @@ -0,0 +1,138 @@ +/* + * 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.distributed.test.accord; + +import java.io.IOException; +import java.util.Arrays; +import java.util.List; + +import org.junit.BeforeClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import accord.local.CommandStore; +import accord.local.CommandStores.PreviouslyOwned; +import accord.local.PreLoadContext; +import accord.primitives.AbstractRanges; +import accord.primitives.Ranges; +import accord.topology.ActiveEpochs; + +import org.apache.cassandra.dht.Murmur3Partitioner; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.api.SimpleQueryResult; +import org.apache.cassandra.schema.Schema; +import org.apache.cassandra.schema.TableId; +import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.service.accord.AccordService; +import org.apache.cassandra.service.accord.TokenRange; +import org.apache.cassandra.service.accord.api.TokenKey; + +import static com.google.common.collect.Iterables.getOnlyElement; +import static org.apache.cassandra.service.accord.AccordService.getBlocking; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public class AccordRegainRangesTest extends AccordTestBase +{ + private static final Logger logger = LoggerFactory.getLogger(AccordRegainRangesTest.class); + + @Override + protected Logger logger() + { + return logger; + } + + @BeforeClass + public static void setupClass() throws IOException + { + AccordTestBase.setupCluster(builder -> builder + .withoutVNodes() + .withConfig(config -> + config + .set("accord.shard_durability_target_splits", "1") + .set("accord.shard_durability_cycle", "20s") + .with(Feature.NETWORK, Feature.GOSSIP)), 6); + } + + @Test + public void regainRangesTest() throws Throwable + { + List ddls = Arrays.asList("DROP KEYSPACE IF EXISTS " + KEYSPACE + ';', + "CREATE KEYSPACE " + KEYSPACE + " WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor': 3}", + "CREATE TABLE " + qualifiedAccordTableName + " (k int PRIMARY KEY, v int) WITH transactional_mode='full'"); + test(ddls, cluster -> { + cluster.coordinator(2).execute(wrapInTxn("INSERT INTO " + qualifiedAccordTableName + " (k, v) VALUES (?, ?)"), ConsistencyLevel.SERIAL, 1, 2); + + SimpleQueryResult result = cluster.coordinator(2).executeWithResult("SELECT token(k) FROM " + qualifiedAccordTableName + " WHERE k = 1 LIMIT 1", ConsistencyLevel.SERIAL); + + String originalToken = cluster.get(2).callOnInstance(() -> getOnlyElement(StorageService.instance.getTokens())); + + long token = (Long) result.toObjectArrays()[0][0]; + + assertTrue(token < Long.parseLong(originalToken)); + + // Loses range (-4069959284402364209,-3074457345618258603] + cluster.get(2).runOnInstance(() -> { + StorageService.instance.move(Long.toString(token - 1000)); + }); + + // Regains range (-4069959284402364209,-3074457345618258603] + cluster.get(2).runOnInstance(() -> { + StorageService.instance.move(originalToken); + }); + + String tableName = accordTableName; + + // Ensure no overlapping safeToRead ranges + cluster.get(2).runOnInstance(() -> { + TableId tid = Schema.instance.getTableMetadata(KEYSPACE, tableName).id(); + TokenKey start = TokenKey.parse(tid, String.valueOf(token), Murmur3Partitioner.instance); + TokenKey end = TokenKey.parse(tid, originalToken, Murmur3Partitioner.instance); + Ranges regainedRange = Ranges.of(TokenRange.create(start, end)); + + ActiveEpochs activeEpochs = AccordService.instance().topology().active(); + PreviouslyOwned previouslyOwned = AccordService.instance().node().commandStores().getPreviouslyOwnedFromSnapshot(); + assertEquals(previouslyOwned.regains(regainedRange), regainedRange); + for (int i = 0; i < previouslyOwned.size(); i++) + { + long epoch = previouslyOwned.epochs(i); + if (!activeEpochs.hasAtLeastEpoch(epoch)) + continue; + assertTrue(activeEpochs.getKnown(epoch).retired().containsAll(previouslyOwned.ranges(i))); + } + + Ranges range = Ranges.EMPTY; + for (CommandStore commandStore : AccordService.instance().node().commandStores().all()) { + Ranges safeToReadRanges = getBlocking(commandStore.submit((PreLoadContext.Empty) () -> "No overlapping safeToReadRanges", safeCommandStore -> { + Ranges mergedRanges = Ranges.EMPTY; + for (Ranges r : safeCommandStore.safeToReadAt().values()) + mergedRanges = mergedRanges.union(AbstractRanges.UnionMode.MERGE_ADJACENT, r); + return mergedRanges; + })); + + assertTrue(range.overlapping(safeToReadRanges).isEmpty()); + range = range.union(AbstractRanges.UnionMode.MERGE_ADJACENT, safeToReadRanges); + } + }); + }); + } +} + diff --git a/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java b/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java index a5d5e2c431..e12ec58752 100644 --- a/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java +++ b/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java @@ -92,10 +92,12 @@ public class AccordJournalCompactionTest NavigableMap safeToReadAtAccumulator = ImmutableSortedMap.of(Timestamp.NONE, Ranges.EMPTY); NavigableMap bootstrapBeganAtAccumulator = ImmutableSortedMap.of(TxnId.NONE, Ranges.EMPTY); RangesForEpoch rangesForEpochAccumulator = null; + Ranges permanentlyUnsafeToReadAccumulator = Ranges.EMPTY; Gen durableBeforeGen = AccordGenerators.durableBeforeGen(DatabaseDescriptor.getPartitioner()); Gen> safeToReadGen = AccordGenerators.safeToReadGen(DatabaseDescriptor.getPartitioner()); Gen rangesForEpochGen = AccordGenerators.rangesForEpoch(DatabaseDescriptor.getPartitioner()); + Gen permanentlyUnsafeToRead = AccordGenerators.ranges(DatabaseDescriptor.getPartitioner()); AccordJournal journal = new AccordJournal(new TestParams() { @@ -129,6 +131,7 @@ public class AccordJournalCompactionTest // updates.addRedundantBefore = redundantBeforeGen.next(rs); updates.newSafeToRead = safeToReadGen.next(rs); updates.newRangesForEpoch = rangesForEpochGen.next(rs); + updates.newPermanentlyUnsafeToRead = permanentlyUnsafeToRead.next(rs); journal.durableBeforePersister().persist(addDurableBefore, null); journal.saveStoreState(1, updates, null); @@ -141,6 +144,8 @@ public class AccordJournalCompactionTest safeToReadAtAccumulator = updates.newSafeToRead; if (updates.newRangesForEpoch != null) rangesForEpochAccumulator = updates.newRangesForEpoch; + if (updates.newPermanentlyUnsafeToRead != null) + permanentlyUnsafeToReadAccumulator = updates.newPermanentlyUnsafeToRead; if (i % 100 == 0) journal.closeCurrentSegmentForTestingIfNonEmpty(); @@ -153,6 +158,7 @@ public class AccordJournalCompactionTest Assert.assertEquals(bootstrapBeganAtAccumulator, journal.loadBootstrapBeganAt(1)); Assert.assertEquals(safeToReadAtAccumulator, journal.loadSafeToRead(1)); Assert.assertEquals(rangesForEpochAccumulator, journal.loadRangesForEpoch(1)); + Assert.assertEquals(permanentlyUnsafeToReadAccumulator, journal.loadPermanentlyUnsafeToRead(1)); } finally { diff --git a/test/unit/org/apache/cassandra/service/accord/journal/TopologyRecordTest.java b/test/unit/org/apache/cassandra/service/accord/journal/TopologyRecordTest.java index 199e7f20db..a3dee6de81 100644 --- a/test/unit/org/apache/cassandra/service/accord/journal/TopologyRecordTest.java +++ b/test/unit/org/apache/cassandra/service/accord/journal/TopologyRecordTest.java @@ -40,6 +40,7 @@ import org.apache.cassandra.schema.TableId; import org.apache.cassandra.utils.AccordGenerators; import static accord.utils.Property.qt; +import static org.apache.cassandra.service.accord.serializers.CommandStoreSerializersTest.previouslyOwnedGen; import static org.apache.cassandra.service.accord.serializers.CommandStoreSerializersTest.rangesForEpochGen; public class TopologyRecordTest @@ -85,14 +86,13 @@ public class TopologyRecordTest Gen rangesGen = AccordGenerators.ranges(TBL1, partitioner); Gen rangesForEpochGen = rangesForEpochGen(rangesGen); Topology topology = AccordGenerators.topologyGen(rangesGen).next(rs); + CommandStores.PreviouslyOwned previouslyOwned = previouslyOwnedGen(rangesGen).next(rs); Int2ObjectHashMap commandStores = new Int2ObjectHashMap<>(); for (Node.Id node : topology.nodes()) commandStores.put(node.id, rangesForEpochGen.next(rs)); - Node.Id self = rs.pick(topology.nodes()); - - return new Journal.TopologyUpdate(commandStores, topology); + return new Journal.TopologyUpdate(commandStores, topology, previouslyOwned); }; } @@ -120,8 +120,8 @@ public class TopologyRecordTest private static void maybeUpdatePartitioner(TopologyRecord expected) { - Journal.TopologyUpdate update = expected.getUpdate(); + Journal.TopologyUpdate update = expected.update(); if (update != null) - maybeUpdatePartitioner(expected.getUpdate()); + maybeUpdatePartitioner(expected.update()); } } \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/CommandStoreSerializersTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/CommandStoreSerializersTest.java index 146c3abc56..af699f591e 100644 --- a/test/unit/org/apache/cassandra/service/accord/serializers/CommandStoreSerializersTest.java +++ b/test/unit/org/apache/cassandra/service/accord/serializers/CommandStoreSerializersTest.java @@ -127,6 +127,27 @@ public class CommandStoreSerializersTest }; } + public static Gen previouslyOwnedGen(Gen rangesGen) + { + Gen.IntGen sizeGen = Gens.ints().between(0, 10); + Gen.LongGen epochGen = AccordGens.epochs(); + return rs -> { + int size = sizeGen.nextInt(rs); + if (size == 0) + return CommandStores.PreviouslyOwned.EMPTY; + long maxEpoch = 0; + long[] epochs = new long[size]; + Ranges[] ranges = new Ranges[size]; + for (int i = 0; i < size; i++) + { + epochs[i] = epochGen.nextLong(rs); + ranges[i] = rangesGen.next(rs); + maxEpoch = Math.max(maxEpoch, epochs[i]); + } + return new CommandStores.PreviouslyOwned(maxEpoch, epochs, ranges); + }; + } + private void maybeUpdatePartitioner(CommandStores.RangesForEpoch expected) { if (expected.size() > 0)