This is an automated email from the ASF dual-hosted git repository.
iamaleksey pushed a commit to branch 20386
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/20386 by this push:
new 33a32e3725 Add an explicit UNLOCK TCM step
33a32e3725 is described below
commit 33a32e372563489e4ac172db6b39d5ffb10d16ae
Author: Aleksey Yeshchenko <[email protected]>
AuthorDate: Mon Aug 24 13:15:05 2026 +0100
Add an explicit UNLOCK TCM step
Allow shard sealing after FINISH but while still having the ranges locked
---
.../cassandra/replication/SealingCoordinator.java | 94 +++++++++++-----
.../org/apache/cassandra/replication/Shard.java | 3 +-
.../org/apache/cassandra/tcm/Transformation.java | 9 ++
.../cassandra/tcm/sequences/BootstrapAndJoin.java | 44 +++++++-
.../tcm/sequences/BootstrapAndReplace.java | 53 +++++++--
.../org/apache/cassandra/tcm/sequences/Move.java | 34 +++++-
.../tcm/sequences/UnbootstrapAndLeave.java | 54 +++++++---
.../cassandra/tcm/serialization/Version.java | 2 +
.../tcm/transformations/ApplyPlacementDeltas.java | 5 +
.../cassandra/tcm/transformations/PrepareJoin.java | 32 ++++--
.../tcm/transformations/PrepareLeave.java | 48 ++++++++-
.../cassandra/tcm/transformations/PrepareMove.java | 30 ++++--
.../tcm/transformations/PrepareReplace.java | 54 +++++++++-
.../tcm/transformations/UnlockSequence.java | 120 +++++++++++++++++++++
.../cassandra/distributed/test/MoveTest.java | 2 +-
.../test/log/ClusterMetadataTestHelper.java | 16 +++
.../distributed/test/log/MetadataKeysTest.java | 17 +++
.../cassandra/simulator/cluster/OnClusterJoin.java | 7 ++
.../simulator/cluster/OnClusterLeave.java | 7 ++
.../simulator/cluster/OnClusterReplace.java | 7 ++
.../tcm/transformations/EventsMetadataTest.java | 3 +
21 files changed, 563 insertions(+), 78 deletions(-)
diff --git a/src/java/org/apache/cassandra/replication/SealingCoordinator.java
b/src/java/org/apache/cassandra/replication/SealingCoordinator.java
index 994c5c771d..cc19be245a 100644
--- a/src/java/org/apache/cassandra/replication/SealingCoordinator.java
+++ b/src/java/org/apache/cassandra/replication/SealingCoordinator.java
@@ -145,17 +145,31 @@ public final class SealingCoordinator
* 5. FINISH_JOIN at e4:
*
* (10,20] | {n20,n30,n40} | e0 | SEALED
- * S (10,20] | {n20,n30,n35,n40} | e2 | SEALED
+ * (10,20] | {n20,n30,n35,n40} | e2 | ACTIVE
* + (10,20] | {n20,n30,n35} | e4 | ACTIVE
* (20,30] | {n30,n40,n50} | e0 | SEALED
- * S (20,30] | {n30,n35,n40,n50} | e2 | SEALED
+ * (20,30] | {n30,n35,n40,n50} | e2 | ACTIVE
* + (20,30] | {n30,n35,n40} | e4 | ACTIVE
* (30,40] | {n40,n50,n60} | e0 | SEALED
* (30,35] | {n40,n50,n60} | e1 | SEALED
- * S (30,35] | {n35,n40,n50,n60} | e2 | SEALED
+ * (30,35] | {n35,n40,n50,n60} | e2 | ACTIVE
* + (30,35] | {n35,n40,n50} | e4 | ACTIVE
* (35,40] | {n40,n50,n60} | e1 | ACTIVE
*
+ * 6. UNLOCK_SEQUENCE:
+ *
+ * (10,20] | {n20,n30,n40} | e0 | SEALED
+ * S (10,20] | {n20,n30,n35,n40} | e2 | SEALED
+ * (10,20] | {n20,n30,n35} | e4 | ACTIVE
+ * (20,30] | {n30,n40,n50} | e0 | SEALED
+ * S (20,30] | {n30,n35,n40,n50} | e2 | SEALED
+ * (20,30] | {n30,n35,n40} | e4 | ACTIVE
+ * (30,40] | {n40,n50,n60} | e0 | SEALED
+ * (30,35] | {n40,n50,n60} | e1 | SEALED
+ * S (30,35] | {n35,n40,n50,n60} | e2 | SEALED
+ * (30,35] | {n35,n40,n50} | e4 | ACTIVE
+ * (35,40] | {n40,n50,n60} | e1 | ACTIVE
+ *
* [*] At this stage, after sealing the shards, we stream the sstables.
* All the offsets belonging to these sealed shards get stripped from
* stats component of streamed sstables. Otherwise, the new node
@@ -172,16 +186,11 @@ public final class SealingCoordinator
}
/**
- * Seal the intermediate shards that were created during START_JOIN.
- * TODO (required): this MUST eventually run while the affected ranges are
still locked - between the
- * removeFromWrites placement change and the range unlock.
FinishJoin/FinishReplace currently apply
- * removeFromWrites and the unlock atomically in a single
transformation, so there is no committed
- * dropped-but-locked epoch to hook. Until then, this runs just after
FINISH_JOIN/FINISH_REPLACE,
- * when the ranges are already unlocked. Left to implement for the
unhappy path (AY).
+ * Seal the intermediate, over-replicated shards created during START_JOIN
and obsoleted by FINISH_JOIN.
*/
- public static void sealShardsAtFinishJoin(ClusterMetadata metadata,
PlacementDeltas finishDelta)
+ public static void sealShardsAtFinishJoin(ClusterMetadata metadata, long
finishEpoch, PlacementDeltas finishDelta)
{
- seal(discoverShardsAtFinishJoinOrReplace(metadata, finishDelta, null));
+ seal(discoverShardsAtFinishJoinOrReplace(metadata, finishEpoch,
finishDelta, null));
}
/*
@@ -219,15 +228,27 @@ public final class SealingCoordinator
* 4. FINISH_REPLACE at e3:
*
* (10,20] | {n20,n30,n40} | e0 | SEALED
- * S (10,20] | {n20,n30,n40,$n40}| e1 | SEALED
+ * (10,20] | {n20,n30,n40,$n40}| e1 | ACTIVE
* + (10,20] | {n20,n30,$n40} | e3 | ACTIVE
* (20,30] | {n30,n40,n50} | e0 | SEALED
- * S (20,30] | {n30,n40,$n40,n50}| e1 | SEALED
+ * (20,30] | {n30,n40,$n40,n50}| e1 | ACTIVE
* + (20,30] | {n30,$n40,n50} | e3 | ACTIVE
* (30,40] | {n40,n50,n60} | e0 | SEALED
- * S (30,40] | {n40,$n40,n50,n60}| e1 | SEALED
+ * (30,40] | {n40,$n40,n50,n60}| e1 | ACTIVE
* + (30,40] | {$n40,n50,n60} | e3 | ACTIVE
*
+ * 5. UNLOCK_SEQUENCE:
+ *
+ * (10,20] | {n20,n30,n40} | e0 | SEALED
+ * S (10,20] | {n20,n30,n40,$n40}| e1 | SEALED
+ * (10,20] | {n20,n30,$n40} | e3 | ACTIVE
+ * (20,30] | {n30,n40,n50} | e0 | SEALED
+ * S (20,30] | {n30,n40,$n40,n50}| e1 | SEALED
+ * (20,30] | {n30,$n40,n50} | e3 | ACTIVE
+ * (30,40] | {n40,n50,n60} | e0 | SEALED
+ * S (30,40] | {n40,$n40,n50,n60}| e1 | SEALED
+ * (30,40] | {$n40,n50,n60} | e3 | ACTIVE
+ *
* [*] At this stage, after sealing the shards, we stream the sstables.
* Non-repaired (not fully reconciled) SSTables must be split into
* fully reconciled part and non-reconciled part. When streaming the
@@ -247,9 +268,13 @@ public final class SealingCoordinator
seal(discoverShardsAtMidJoinOrReplace(metadata, movements), deadNode);
}
- public static void sealShardsAtFinishReplace(ClusterMetadata metadata,
PlacementDeltas finishDelta, @Nonnull NodeId deadNode)
+ /**
+ * Seal the intermediate, over-replicated shards created during
START_REPLACE and obsoleted by FINISH_REPLACE.
+ * The {@code deadNode} node cannot participate in the sealing.
+ */
+ public static void sealShardsAtFinishReplace(ClusterMetadata metadata,
long finishEpoch, PlacementDeltas finishDelta, @Nonnull NodeId deadNode)
{
- seal(discoverShardsAtFinishJoinOrReplace(metadata, finishDelta,
deadNode), deadNode);
+ seal(discoverShardsAtFinishJoinOrReplace(metadata, finishEpoch,
finishDelta, deadNode), deadNode);
}
/**
@@ -297,9 +322,9 @@ public final class SealingCoordinator
/**
* Discover all shards that need to be sealed at FINISH_JOIN or
FINISH_REPLACE.
*/
- private static Set<ShardMetadata>
discoverShardsAtFinishJoinOrReplace(ClusterMetadata metadata, PlacementDeltas
finishDelta, @Nullable NodeId deadNode)
+ private static Set<ShardMetadata> discoverShardsAtFinishJoinOrReplace(
+ ClusterMetadata metadata, long finishEpoch, PlacementDeltas
finishDelta, @Nullable NodeId deadNode)
{
- long finishEpoch = metadata.epoch.getEpoch();
Set<ShardMetadata> shards = new HashSet<>();
for (KeyspaceMetadata ksm : metadata.schema.getKeyspaces())
{
@@ -364,15 +389,28 @@ public final class SealingCoordinator
* 4. FINISH_LEAVE at e3:
*
* (10,20] | {n20,n30,n40} | e0 | SEALED
- * S (10,20] | {n20,n30,n40,n50} | e1 | SEALED
+ * (10,20] | {n20,n30,n40,n50} | e1 | ACTIVE
* + (10,20] | {n20,n30,n50} | e3 | ACTIVE
* (20,30] | {n30,n40,n50} | e0 | SEALED
- * S (20,30] | {n30,n40,n50,n60} | e1 | SEALED
+ * (20,30] | {n30,n40,n50,n60} | e1 | ACTIVE
* + (20,30] | {n30,n50,n60} | e3 | ACTIVE
* (30,40] | {n40,n50,n60} | e0 | SEALED
+ * (30,40] | {n40,n50,n60,n70} | e1 | ACTIVE
+ * (40,50] | {n50,n60,n70} | e0 | ACTIVE
+ * + (30,50] | {n50,n60,n70} | e3 | ACTIVE
+ *
+ * 5. UNLOCK_SEQUENCE:
+ *
+ * (10,20] | {n20,n30,n40} | e0 | SEALED
+ * S (10,20] | {n20,n30,n40,n50} | e1 | SEALED
+ * (10,20] | {n20,n30,n50} | e3 | ACTIVE
+ * (20,30] | {n30,n40,n50} | e0 | SEALED
+ * S (20,30] | {n30,n40,n50,n60} | e1 | SEALED
+ * (20,30] | {n30,n50,n60} | e3 | ACTIVE
+ * (30,40] | {n40,n50,n60} | e0 | SEALED
* S (30,40] | {n40,n50,n60,n70} | e1 | SEALED
* S (40,50] | {n50,n60,n70} | e0 | SEALED
- * + (30,50] | {n50,n60,n70} | e3 | ACTIVE
+ * (30,50] | {n50,n60,n70} | e3 | ACTIVE
*
* [*] At this stage, after sealing the shards, we stream the sstables.
* Non-repaired (not fully reconciled) SSTables must be split into
@@ -424,20 +462,20 @@ public final class SealingCoordinator
}
/**
- * Seal the shards that need to be sealed at LEAVE_FINISH for both
UNBOOTSTRAP and REMOVENODE.
+ * Seal the shards obsoleted by FINISH_LEAVE, for both UNBOOTSTRAP and
REMOVENODE: the departed node's
+ * over-replicated generations created during START_LEAVE, plus the
merge-half folded into its range.
*/
- public static void sealShardsAtFinishLeave(ClusterMetadata postFinish,
PlacementDeltas finishDelta, NodeId leavingOrDeparted, LeaveStreams.Kind kind)
+ public static void sealShardsAtFinishLeave(ClusterMetadata metadata, long
finishEpoch, PlacementDeltas finishDelta, NodeId leavingOrDeparted,
LeaveStreams.Kind kind)
{
- long finishEpoch = postFinish.epoch.getEpoch();
- InetAddressAndPort leavingOrDepartedEndpoint =
postFinish.directory.endpoint(leavingOrDeparted);
+ InetAddressAndPort leavingOrDepartedEndpoint =
metadata.directory.endpoint(leavingOrDeparted);
List<Future<Set<ShardMetadata>>> futures = new ArrayList<>();
- for (KeyspaceMetadata ksm : postFinish.schema.getKeyspaces())
+ for (KeyspaceMetadata ksm : metadata.schema.getKeyspaces())
{
if (!ksm.params.replicationType.isTracked())
continue;
- DataPlacement placement =
postFinish.placements.get(ksm.params.replication);
+ DataPlacement placement =
metadata.placements.get(ksm.params.replication);
RangesAtEndpoint removals =
finishDelta.get(ksm.params.replication).writes.removals.get(leavingOrDepartedEndpoint);
for (Replica removedReplica : removals)
{
@@ -507,7 +545,7 @@ public final class SealingCoordinator
* it no longer allocates new ids on it). The ProgressBarrier gating
bootstrap()/MID_JOIN only waits for
* EACH_QUORUM of the affected replicas, NOT all live replicas, so a
live-but-lagging participant could still
* be on the old shard. Bringing each participant up to the seal epoch
must be done per-replica (not
- * per-shard) and will be added separately. TLDR - add a step to
guaranteed up-to-speed rest of replicas epochs.
+ * per-shard) and will be added later.
*/
private static AsyncPromise<Void> initiate(ShardMetadata shard, @Nullable
NodeId withoutNode)
{
diff --git a/src/java/org/apache/cassandra/replication/Shard.java
b/src/java/org/apache/cassandra/replication/Shard.java
index e2b22f653c..3e345e3d98 100644
--- a/src/java/org/apache/cassandra/replication/Shard.java
+++ b/src/java/org/apache/cassandra/replication/Shard.java
@@ -471,8 +471,9 @@ public class Shard
void markSealing()
{
+ // TODO (expected): consider what can possibly go wrong with the no-op
on SEALED
if (state == State.SEALED)
- throw new IllegalStateException(format("%s cannot transition to
SEALING from %s", this, state));
+ return;
if (state != State.SEALING)
{
diff --git a/src/java/org/apache/cassandra/tcm/Transformation.java
b/src/java/org/apache/cassandra/tcm/Transformation.java
index 664cf5a6a8..9593386c9a 100644
--- a/src/java/org/apache/cassandra/tcm/Transformation.java
+++ b/src/java/org/apache/cassandra/tcm/Transformation.java
@@ -60,6 +60,7 @@ import
org.apache.cassandra.tcm.transformations.ReconfigureAccordFastPath;
import org.apache.cassandra.tcm.transformations.Register;
import org.apache.cassandra.tcm.transformations.Startup;
import org.apache.cassandra.tcm.transformations.TriggerSnapshot;
+import org.apache.cassandra.tcm.transformations.UnlockSequence;
import org.apache.cassandra.tcm.transformations.Unregister;
import org.apache.cassandra.tcm.transformations.UnsafeJoin;
import org.apache.cassandra.tcm.transformations.cms.AdvanceCMSReconfiguration;
@@ -275,6 +276,14 @@ public interface Transformation
FINISH_DROP_ACCORD_TABLE(42, Version.MIN_ACCORD_VERSION, () ->
FinishDropAccordTable.serializer),
ACCORD_MARK_HARD_REMOVED(43, Version.MIN_ACCORD_VERSION, () ->
AccordMarkHardRemoved.serializer),
ADVANCE_MUTATION_TRACKING_MIGRATION(44,
Version.MIN_MUTATION_TRACKING_VERSION, () ->
AdvanceMutationTrackingMigration.serializer),
+
+ /*
+ * Additional final step for JOIN/MOVE/LEAVE/REPLACE sequences that
seal the
+ * intermediate mutation tracking shards before unlocking the affected
ranges.
+ * See o.a.c.replication.SealingCoordinator comments for additional
context.
+ * TODO (now): what should be the correct introducedIn version for
these?
+ */
+ UNLOCK_SEQUENCE(45, Version.TBD_UNLOCK_STEP, () ->
UnlockSequence.serializer),
;
/**
diff --git a/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java
b/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java
index 455953cc42..cbf9724618 100644
--- a/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java
+++ b/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java
@@ -70,6 +70,7 @@ import
org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer;
import org.apache.cassandra.tcm.serialization.MetadataSerializer;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.tcm.transformations.PrepareJoin;
+import org.apache.cassandra.tcm.transformations.UnlockSequence;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.concurrent.Future;
@@ -82,6 +83,7 @@ import static
org.apache.cassandra.tcm.MultiStepOperation.Kind.JOIN;
import static org.apache.cassandra.tcm.Transformation.Kind.FINISH_JOIN;
import static org.apache.cassandra.tcm.Transformation.Kind.MID_JOIN;
import static org.apache.cassandra.tcm.Transformation.Kind.START_JOIN;
+import static org.apache.cassandra.tcm.Transformation.Kind.UNLOCK_SEQUENCE;
import static org.apache.cassandra.tcm.sequences.SequenceState.continuable;
import static org.apache.cassandra.tcm.sequences.SequenceState.error;
import static org.apache.cassandra.tcm.sequences.SequenceState.halted;
@@ -186,7 +188,15 @@ public class BootstrapAndJoin extends
MultiStepOperation<Epoch>
@Override
public Transformation.Result applyTo(ClusterMetadata metadata)
{
- return applyMultipleTransformations(metadata, next, of(startJoin,
midJoin, finishJoin));
+ // FinishJoin unlocks the affected ranges and retires the sequence
only in legacy (pre-UNLOCK_SEQUENCE) mode.
+ return finishJoin.unlocks()
+ ? applyMultipleTransformations(metadata, next, of(startJoin,
midJoin, finishJoin))
+ : applyMultipleTransformations(metadata, next, of(startJoin,
midJoin, finishJoin, unlockSequence()));
+ }
+
+ private UnlockSequence unlockSequence()
+ {
+ return new UnlockSequence(startJoin.nodeId(), lockKey);
}
@Override
@@ -303,8 +313,24 @@ public class BootstrapAndJoin extends
MultiStepOperation<Epoch>
}
ClusterMetadataService.instance().ensureCMSPlacement(metadata);
- if (MutationTrackingService.isEnabled())
- SealingCoordinator.sealShardsAtFinishJoin(metadata,
finishJoin.delta());
+ break;
+ case UNLOCK_SEQUENCE:
+ try
+ {
+ if (MutationTrackingService.isEnabled())
+ {
+
SealingCoordinator.sealShardsAtFinishJoin(ClusterMetadata.current(),
+
latestModification.getEpoch(),
+
finishJoin.delta());
+ }
+ ClusterMetadataService.instance().commit(unlockSequence());
+ }
+ catch (Throwable e)
+ {
+ JVMStabilityInspector.inspectThrowable(e);
+ logger.warn("Exception sealing obsoleted MT shards or
committing unlockSequence", e);
+ return continuable();
+ }
break;
default:
@@ -333,11 +359,15 @@ public class BootstrapAndJoin extends
MultiStepOperation<Epoch>
@Override
public ClusterMetadata.Transformer cancel(ClusterMetadata metadata)
{
+ // TODO (expected): reason about shard sealing states at various stages
DataPlacements placements = metadata.placements;
switch (next)
{
- // need to undo MID_JOIN and START_JOIN, then merge the ranges
split by PrepareJoin
+ case UNLOCK_SEQUENCE:
+ // FINISH_JOIN has already been enacted - it's too late to be
reverting anything, so we just unlock.
+ return
metadata.transformer().with(metadata.lockedRanges.unlock(lockKey));
case FINISH_JOIN:
+ // need to undo MID_JOIN and START_JOIN, then merge the ranges
split by PrepareJoin
placements =
midJoin.inverseDelta().apply(metadata.nextEpoch(), placements);
case MID_JOIN:
placements =
startJoin.inverseDelta().apply(metadata.nextEpoch(), placements);
@@ -488,6 +518,8 @@ public class BootstrapAndJoin extends
MultiStepOperation<Epoch>
return 1;
case FINISH_JOIN:
return 2;
+ case UNLOCK_SEQUENCE:
+ return 3;
default:
throw new IllegalStateException(String.format("Step %s is
invalid for sequence %s ", next, JOIN));
}
@@ -503,6 +535,8 @@ public class BootstrapAndJoin extends
MultiStepOperation<Epoch>
return MID_JOIN;
case 2:
return FINISH_JOIN;
+ case 3:
+ return UNLOCK_SEQUENCE;
default:
throw new IllegalStateException(String.format("Step %s is
invalid for sequence %s ", index, JOIN));
}
@@ -587,7 +621,7 @@ public class BootstrapAndJoin extends
MultiStepOperation<Epoch>
size += LockedRanges.Key.serializer.serializedSize(plan.lockKey,
version);
size +=
PlacementDeltas.serializer.serializedSize(plan.toSplitRanges, version);
- size += VIntCoding.computeVIntSize(plan.kind().ordinal());
+ size += VIntCoding.computeVIntSize(plan.next.ordinal());
size +=
PrepareJoin.StartJoin.serializer.serializedSize(plan.startJoin, version);
size +=
PrepareJoin.MidJoin.serializer.serializedSize(plan.midJoin, version);
diff --git
a/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndReplace.java
b/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndReplace.java
index 0abbd5d815..9ee52186b5 100644
--- a/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndReplace.java
+++ b/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndReplace.java
@@ -25,6 +25,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
+import java.util.function.Predicate;
import java.util.stream.StreamSupport;
import com.google.common.annotations.VisibleForTesting;
@@ -56,6 +57,7 @@ import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.MultiStepOperation;
import org.apache.cassandra.tcm.Transformation;
import org.apache.cassandra.tcm.membership.Directory;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.membership.NodeState;
import org.apache.cassandra.tcm.ownership.DataPlacement;
@@ -66,6 +68,7 @@ import
org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer;
import org.apache.cassandra.tcm.serialization.MetadataSerializer;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.tcm.transformations.PrepareReplace;
+import org.apache.cassandra.tcm.transformations.UnlockSequence;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.vint.VIntCoding;
@@ -75,6 +78,7 @@ import static
org.apache.cassandra.tcm.MultiStepOperation.Kind.REPLACE;
import static org.apache.cassandra.tcm.Transformation.Kind.FINISH_REPLACE;
import static org.apache.cassandra.tcm.Transformation.Kind.MID_REPLACE;
import static org.apache.cassandra.tcm.Transformation.Kind.START_REPLACE;
+import static org.apache.cassandra.tcm.Transformation.Kind.UNLOCK_SEQUENCE;
import static org.apache.cassandra.tcm.sequences.BootstrapAndJoin.bootstrap;
import static org.apache.cassandra.tcm.sequences.SequenceState.continuable;
import static org.apache.cassandra.tcm.sequences.SequenceState.error;
@@ -180,7 +184,15 @@ public class BootstrapAndReplace extends
MultiStepOperation<Epoch>
@Override
public Transformation.Result applyTo(ClusterMetadata metadata)
{
- return applyMultipleTransformations(metadata, next, of(startReplace,
midReplace, finishReplace));
+ // FinishReplace unlocks the affected ranges and retires the sequence
only in legacy (pre-UNLOCK_SEQUENCE) mode.
+ return finishReplace.unlocks()
+ ? applyMultipleTransformations(metadata, next, of(startReplace,
midReplace, finishReplace))
+ : applyMultipleTransformations(metadata, next, of(startReplace,
midReplace, finishReplace, unlockSequence()));
+ }
+
+ private UnlockSequence unlockSequence()
+ {
+ return new UnlockSequence(startReplace.nodeId(), lockKey);
}
@Override
@@ -292,8 +304,25 @@ public class BootstrapAndReplace extends
MultiStepOperation<Epoch>
}
ClusterMetadataService.instance().ensureCMSPlacement(metadata);
- if (MutationTrackingService.isEnabled())
- SealingCoordinator.sealShardsAtFinishReplace(metadata,
finishReplace.delta(), startReplace.replaced());
+ break;
+ case UNLOCK_SEQUENCE:
+ try
+ {
+ if (MutationTrackingService.isEnabled())
+ {
+
SealingCoordinator.sealShardsAtFinishReplace(ClusterMetadata.current(),
+
latestModification.getEpoch(),
+
finishReplace.delta(),
+
startReplace.replaced());
+ }
+ ClusterMetadataService.instance().commit(unlockSequence());
+ }
+ catch (Throwable e)
+ {
+ JVMStabilityInspector.inspectThrowable(e);
+ logger.warn("Exception sealing obsoleted MT shards or
committing unlockSequence", e);
+ return continuable();
+ }
break;
default:
@@ -316,8 +345,13 @@ public class BootstrapAndReplace extends
MultiStepOperation<Epoch>
if (next == START_REPLACE)
return ProgressBarrier.immediate();
ClusterMetadata metadata = ClusterMetadata.current();
- InetAddressAndPort replaced =
metadata.directory.getNodeAddresses(startReplace.replaced()).broadcastAddress;
- return new ProgressBarrier(latestModification,
metadata.directory.location(startReplace.nodeId()),
metadata.lockedRanges.locked.get(lockKey), e -> !e.equals(replaced));
+ // By UNLOCK_SEQUENCE the replaced node has already been removed from
the directory by FINISH_REPLACE, so there
+ // is nothing left to exclude from the barrier.
+ NodeAddresses replacedAddresses =
metadata.directory.getNodeAddresses(startReplace.replaced());
+ Predicate<InetAddressAndPort> filter = replacedAddresses == null
+ ? e -> true
+ : e ->
!e.equals(replacedAddresses.broadcastAddress);
+ return new ProgressBarrier(latestModification,
metadata.directory.location(startReplace.nodeId()),
metadata.lockedRanges.locked.get(lockKey), filter);
}
@Override
@@ -326,6 +360,9 @@ public class BootstrapAndReplace extends
MultiStepOperation<Epoch>
DataPlacements placements = metadata.placements;
switch (next)
{
+ case UNLOCK_SEQUENCE:
+ // FINISH_REPLACE has already been enacted - it's too late to
be reverting anything, so we just unlock.
+ return
metadata.transformer().with(metadata.lockedRanges.unlock(lockKey));
// need to undo MID_REPLACE and START_REPLACE, but PREPARE_REPLACE
doesn't affect placements
case FINISH_REPLACE:
placements =
midReplace.inverseDelta().apply(metadata.nextEpoch(), placements);
@@ -385,6 +422,8 @@ public class BootstrapAndReplace extends
MultiStepOperation<Epoch>
return 1;
case FINISH_REPLACE:
return 2;
+ case UNLOCK_SEQUENCE:
+ return 3;
default:
throw new IllegalStateException(String.format("Step %s is
invalid for sequence %s ", next, REPLACE));
}
@@ -400,6 +439,8 @@ public class BootstrapAndReplace extends
MultiStepOperation<Epoch>
return MID_REPLACE;
case 2:
return FINISH_REPLACE;
+ case 3:
+ return UNLOCK_SEQUENCE;
default:
throw new IllegalStateException(String.format("Step %s is
invalid for sequence %s ", index, REPLACE));
}
@@ -531,7 +572,7 @@ public class BootstrapAndReplace extends
MultiStepOperation<Epoch>
size += Epoch.serializer.serializedSize(plan.latestModification,
version);
size += LockedRanges.Key.serializer.serializedSize(plan.lockKey,
version);
- size += VIntCoding.computeVIntSize(plan.kind().ordinal());
+ size += VIntCoding.computeVIntSize(plan.next.ordinal());
size +=
PrepareReplace.StartReplace.serializer.serializedSize(plan.startReplace,
version);
size +=
PrepareReplace.MidReplace.serializer.serializedSize(plan.midReplace, version);
diff --git a/src/java/org/apache/cassandra/tcm/sequences/Move.java
b/src/java/org/apache/cassandra/tcm/sequences/Move.java
index a7f316e38f..d33db1b8c1 100644
--- a/src/java/org/apache/cassandra/tcm/sequences/Move.java
+++ b/src/java/org/apache/cassandra/tcm/sequences/Move.java
@@ -73,6 +73,7 @@ import
org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer;
import org.apache.cassandra.tcm.serialization.MetadataSerializer;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.tcm.transformations.PrepareMove;
+import org.apache.cassandra.tcm.transformations.UnlockSequence;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.concurrent.Future;
@@ -84,6 +85,7 @@ import static
org.apache.cassandra.tcm.MultiStepOperation.Kind.MOVE;
import static org.apache.cassandra.tcm.Transformation.Kind.FINISH_MOVE;
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.Transformation.Kind.UNLOCK_SEQUENCE;
import static org.apache.cassandra.tcm.sequences.SequenceState.continuable;
import static org.apache.cassandra.tcm.sequences.SequenceState.error;
@@ -193,7 +195,15 @@ public class Move extends MultiStepOperation<Epoch>
@Override
public Transformation.Result applyTo(ClusterMetadata metadata)
{
- return applyMultipleTransformations(metadata, next, of(startMove,
midMove, finishMove));
+ // FinishMove unlocks the affected ranges and retires the sequence
only in legacy (pre-UNLOCK_SEQUENCE) mode.
+ return finishMove.unlocks()
+ ? applyMultipleTransformations(metadata, next, of(startMove,
midMove, finishMove))
+ : applyMultipleTransformations(metadata, next, of(startMove,
midMove, finishMove, unlockSequence()));
+ }
+
+ private UnlockSequence unlockSequence()
+ {
+ return new UnlockSequence(startMove.nodeId(), lockKey);
}
@Override
@@ -317,6 +327,19 @@ public class Move extends MultiStepOperation<Epoch>
}
ClusterMetadataService.instance().ensureCMSPlacement(metadata);
break;
+ case UNLOCK_SEQUENCE:
+ try
+ {
+ // TODO (required): mutation tracking shard sealing for
move is not yet implemented
+ ClusterMetadataService.instance().commit(unlockSequence());
+ }
+ catch (Throwable t)
+ {
+ JVMStabilityInspector.inspectThrowable(t);
+ logger.warn("Exception committing unlockSequence", t);
+ return continuable();
+ }
+ break;
default:
return error(new IllegalStateException("Can't proceed with
join from " + next));
}
@@ -346,6 +369,9 @@ public class Move extends MultiStepOperation<Epoch>
switch (next)
{
+ case UNLOCK_SEQUENCE:
+ // FINISH_MOVE has already been enacted - it's too late to be
reverting anything, so we just unlock.
+ return
metadata.transformer().with(metadata.lockedRanges.unlock(lockKey));
case FINISH_MOVE:
placements =
midMove.inverseDelta().apply(metadata.nextEpoch(), placements);
case MID_MOVE:
@@ -490,6 +516,8 @@ public class Move extends MultiStepOperation<Epoch>
return 1;
case FINISH_MOVE:
return 2;
+ case UNLOCK_SEQUENCE:
+ return 3;
default:
throw new IllegalStateException(String.format("Step %s is
invalid for sequence %s ", next, MOVE));
}
@@ -505,6 +533,8 @@ public class Move extends MultiStepOperation<Epoch>
return MID_MOVE;
case 2:
return FINISH_MOVE;
+ case 3:
+ return UNLOCK_SEQUENCE;
default:
throw new IllegalStateException(String.format("Step %s is
invalid for sequence %s ", index, MOVE));
}
@@ -601,7 +631,7 @@ public class Move extends MultiStepOperation<Epoch>
size += LockedRanges.Key.serializer.serializedSize(plan.lockKey,
version);
size +=
PlacementDeltas.serializer.serializedSize(plan.toSplitRanges, version);
- size += VIntCoding.computeVIntSize(plan.kind().ordinal());
+ size += VIntCoding.computeVIntSize(plan.next.ordinal());
size +=
PrepareMove.StartMove.serializer.serializedSize(plan.startMove, version);
size +=
PrepareMove.MidMove.serializer.serializedSize(plan.midMove, version);
diff --git
a/src/java/org/apache/cassandra/tcm/sequences/UnbootstrapAndLeave.java
b/src/java/org/apache/cassandra/tcm/sequences/UnbootstrapAndLeave.java
index e923c44dab..0afb2b718d 100644
--- a/src/java/org/apache/cassandra/tcm/sequences/UnbootstrapAndLeave.java
+++ b/src/java/org/apache/cassandra/tcm/sequences/UnbootstrapAndLeave.java
@@ -51,6 +51,7 @@ import
org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer;
import org.apache.cassandra.tcm.serialization.MetadataSerializer;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.tcm.transformations.PrepareLeave;
+import org.apache.cassandra.tcm.transformations.UnlockSequence;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.vint.VIntCoding;
@@ -59,6 +60,7 @@ import static
org.apache.cassandra.tcm.MultiStepOperation.Kind.LEAVE;
import static org.apache.cassandra.tcm.Transformation.Kind.FINISH_LEAVE;
import static org.apache.cassandra.tcm.Transformation.Kind.MID_LEAVE;
import static org.apache.cassandra.tcm.Transformation.Kind.START_LEAVE;
+import static org.apache.cassandra.tcm.Transformation.Kind.UNLOCK_SEQUENCE;
import static org.apache.cassandra.tcm.sequences.SequenceState.continuable;
import static org.apache.cassandra.tcm.sequences.SequenceState.error;
@@ -159,7 +161,15 @@ public class UnbootstrapAndLeave extends
MultiStepOperation<Epoch>
@Override
public Transformation.Result applyTo(ClusterMetadata metadata)
{
- return applyMultipleTransformations(metadata, next, of(startLeave,
midLeave, finishLeave));
+ // FinishLeave unlocks the affected ranges and retires the sequence
only in legacy (pre-UNLOCK_SEQUENCE) mode.
+ return finishLeave.unlocks()
+ ? applyMultipleTransformations(metadata, next, of(startLeave,
midLeave, finishLeave))
+ : applyMultipleTransformations(metadata, next, of(startLeave,
midLeave, finishLeave, unlockSequence()));
+ }
+
+ private UnlockSequence unlockSequence()
+ {
+ return new UnlockSequence(startLeave.nodeId(), lockKey);
}
@Override
@@ -213,10 +223,9 @@ public class UnbootstrapAndLeave extends
MultiStepOperation<Epoch>
}
break;
case FINISH_LEAVE:
- ClusterMetadata postFinish;
try
{
- postFinish =
ClusterMetadataService.instance().commit(finishLeave);
+ ClusterMetadataService.instance().commit(finishLeave);
StorageService.instance.clearTransientMode();
}
catch (Throwable t)
@@ -224,12 +233,26 @@ public class UnbootstrapAndLeave extends
MultiStepOperation<Epoch>
JVMStabilityInspector.inspectThrowable(t);
return continuable();
}
-
- // Seal the shards obsoleted by FINISH_LEAVE: the departed
node's over-replicated generations created
- // during START_LEAVE, plus the merge-half folded into the
departed node's range by the merge.
- if (MutationTrackingService.isEnabled())
- if (streams.kind() == LeaveStreams.Kind.UNBOOTSTRAP ||
streams.kind() == LeaveStreams.Kind.REMOVENODE)
- SealingCoordinator.sealShardsAtFinishLeave(postFinish,
finishLeave.delta(), finishLeave.nodeId(), streams.kind());
+ break;
+ case UNLOCK_SEQUENCE:
+ try
+ {
+ if (MutationTrackingService.isEnabled() && streams.kind()
!= LeaveStreams.Kind.ASSASSINATE)
+ {
+
SealingCoordinator.sealShardsAtFinishLeave(ClusterMetadata.current(),
+
latestModification.getEpoch(),
+
finishLeave.delta(),
+
finishLeave.nodeId(),
+
streams.kind());
+ }
+ ClusterMetadataService.instance().commit(unlockSequence());
+ }
+ catch (Throwable t)
+ {
+ JVMStabilityInspector.inspectThrowable(t);
+ logger.warn("Exception sealing obsoleted MT shards or
committing unlockSequence", t);
+ return continuable();
+ }
break;
default:
return error(new IllegalStateException("Can't proceed with
leave from " + next));
@@ -262,6 +285,9 @@ public class UnbootstrapAndLeave extends
MultiStepOperation<Epoch>
DataPlacements placements = metadata.placements;
switch (next)
{
+ case UNLOCK_SEQUENCE:
+ // FINISH_LEAVE has already been enacted - it's too late to be
reverting anything, so we just unlock.
+ return
metadata.transformer().with(metadata.lockedRanges.unlock(lockKey));
// need to undo MID_LEAVE and START_LEAVE, but PrepareLeave
doesn't affect placement
case FINISH_LEAVE:
placements =
midLeave.inverseDelta().apply(metadata.nextEpoch(), placements);
@@ -296,6 +322,8 @@ public class UnbootstrapAndLeave extends
MultiStepOperation<Epoch>
return 1;
case FINISH_LEAVE:
return 2;
+ case UNLOCK_SEQUENCE:
+ return 3;
default:
throw new IllegalStateException(String.format("Step %s is
invalid for sequence %s ", next, LEAVE));
}
@@ -311,6 +339,8 @@ public class UnbootstrapAndLeave extends
MultiStepOperation<Epoch>
return MID_LEAVE;
case 2:
return FINISH_LEAVE;
+ case 3:
+ return UNLOCK_SEQUENCE;
default:
throw new IllegalStateException(String.format("Step %s is
invalid for sequence %s ", index, LEAVE));
}
@@ -393,11 +423,11 @@ public class UnbootstrapAndLeave extends
MultiStepOperation<Epoch>
long size =
Epoch.serializer.serializedSize(plan.latestModification, version);
size += LockedRanges.Key.serializer.serializedSize(plan.lockKey,
version);
- size += VIntCoding.computeVIntSize(plan.kind().ordinal());
+ size += VIntCoding.computeVIntSize(plan.next.ordinal());
size += VIntCoding.computeVIntSize(plan.streams.kind().ordinal());
size +=
PrepareLeave.StartLeave.serializer.serializedSize(plan.startLeave, version);
- size +=
PrepareLeave.StartLeave.serializer.serializedSize(plan.midLeave, version);
- size +=
PrepareLeave.StartLeave.serializer.serializedSize(plan.finishLeave, version);
+ size +=
PrepareLeave.MidLeave.serializer.serializedSize(plan.midLeave, version);
+ size +=
PrepareLeave.FinishLeave.serializer.serializedSize(plan.finishLeave, version);
return size;
}
}
diff --git a/src/java/org/apache/cassandra/tcm/serialization/Version.java
b/src/java/org/apache/cassandra/tcm/serialization/Version.java
index 6f0e72815f..119399bf61 100644
--- a/src/java/org/apache/cassandra/tcm/serialization/Version.java
+++ b/src/java/org/apache/cassandra/tcm/serialization/Version.java
@@ -90,7 +90,9 @@ public enum Version
* The version that Accord was added to TCM.
*/
public static final Version MIN_ACCORD_VERSION = V7;
+
public static final Version MIN_MUTATION_TRACKING_VERSION = V8;
+ public static final Version TBD_UNLOCK_STEP = V9;
private static Map<Integer, Version> values = new HashMap<>();
diff --git
a/src/java/org/apache/cassandra/tcm/transformations/ApplyPlacementDeltas.java
b/src/java/org/apache/cassandra/tcm/transformations/ApplyPlacementDeltas.java
index 10f48609d4..e893710954 100644
---
a/src/java/org/apache/cassandra/tcm/transformations/ApplyPlacementDeltas.java
+++
b/src/java/org/apache/cassandra/tcm/transformations/ApplyPlacementDeltas.java
@@ -63,6 +63,11 @@ public abstract class ApplyPlacementDeltas implements
Transformation
return delta;
}
+ public boolean unlocks()
+ {
+ return unlock;
+ }
+
public Set<InetAddressAndPort> affectedEndpoints()
{
Set<InetAddressAndPort> affectedEndpoints = new HashSet<>();
diff --git a/src/java/org/apache/cassandra/tcm/transformations/PrepareJoin.java
b/src/java/org/apache/cassandra/tcm/transformations/PrepareJoin.java
index d8986e6829..065fe9d74f 100644
--- a/src/java/org/apache/cassandra/tcm/transformations/PrepareJoin.java
+++ b/src/java/org/apache/cassandra/tcm/transformations/PrepareJoin.java
@@ -160,7 +160,8 @@ public class PrepareJoin implements Transformation
LockedRanges.Key lockKey = LockedRanges.keyFor(prev.nextEpoch());
StartJoin startJoin = new StartJoin(nodeId,
transitionPlan.addToWrites(), lockKey);
MidJoin midJoin = new MidJoin(nodeId, transitionPlan.moveReads(),
lockKey);
- FinishJoin finishJoin = new FinishJoin(nodeId, tokens,
transitionPlan.removeFromWrites(), lockKey);
+ FinishJoin finishJoin =
+ new FinishJoin(nodeId, tokens, transitionPlan.removeFromWrites(),
lockKey, !UnlockSequence.isSupportedBy(prev));
BootstrapAndJoin plan = BootstrapAndJoin.newSequence(prev.nextEpoch(),
lockKey,
@@ -308,7 +309,12 @@ public class PrepareJoin implements Transformation
public FinishJoin(NodeId nodeId, Set<Token> tokens, PlacementDeltas
delta, LockedRanges.Key unlockKey)
{
- super(nodeId, delta, unlockKey, true);
+ this(nodeId, tokens, delta, unlockKey, true);
+ }
+
+ public FinishJoin(NodeId nodeId, Set<Token> tokens, PlacementDeltas
delta, LockedRanges.Key unlockKey, boolean unlock)
+ {
+ super(nodeId, delta, unlockKey, unlock);
this.tokens = tokens;
}
@@ -321,10 +327,12 @@ public class PrepareJoin implements Transformation
@Override
public ClusterMetadata.Transformer transform(ClusterMetadata prev,
ClusterMetadata.Transformer transformer)
{
- return transformer.join(nodeId)
- .proposeToken(nodeId, tokens)
- .addToRackAndDC(nodeId)
- .with(prev.inProgressSequences.without(nodeId));
+ ClusterMetadata.Transformer next =
+ transformer.join(nodeId)
+ .proposeToken(nodeId, tokens)
+ .addToRackAndDC(nodeId);
+ return unlock ? next.with(prev.inProgressSequences.without(nodeId))
+ : next.with(prev.inProgressSequences.with(nodeId,
(BootstrapAndJoin plan) -> plan.advance(prev.nextEpoch())));
}
public static final class Serializer extends SerializerBase<FinishJoin>
@@ -337,6 +345,9 @@ public class PrepareJoin implements Transformation
out.writeUnsignedVInt32(tokens.size());
for (Token token : tokens)
Token.metadataSerializer.serialize(token, out, version);
+ // TODO (now): confirm that this is the Version to use
+ if (version.isAtLeast(Version.TBD_UNLOCK_STEP))
+ out.writeBoolean(((FinishJoin) t).unlock);
}
@Override
@@ -350,7 +361,11 @@ public class PrepareJoin implements Transformation
IPartitioner partitioner =
ClusterMetadata.current().partitioner;
for (int i = 0; i < numTokens; i++)
tokens.add(Token.metadataSerializer.deserialize(in,
partitioner, version));
- return new FinishJoin(nodeId, tokens, delta, lockKey);
+ // UnlockSequence step which follows after FinishJoin was
introduced in TBD_UNLOCK_STEP
+ // to support mutation tracking, and unlock now happens in
that step.
+ // TODO (now): confirm the correct Version to use here.
+ boolean unlock = version.isBefore(Version.TBD_UNLOCK_STEP) ||
in.readBoolean();
+ return new FinishJoin(nodeId, tokens, delta, lockKey, unlock);
}
@Override
@@ -361,6 +376,9 @@ public class PrepareJoin implements Transformation
size += TypeSizes.sizeofUnsignedVInt(tokens.size());
for (Token token : tokens)
size += Token.metadataSerializer.serializedSize(token,
version);
+ // TODO (now): confirm the correct Version to use here.
+ if (version.isAtLeast(Version.TBD_UNLOCK_STEP))
+ size += TypeSizes.sizeof(((FinishJoin)t).unlock);
return size;
}
diff --git
a/src/java/org/apache/cassandra/tcm/transformations/PrepareLeave.java
b/src/java/org/apache/cassandra/tcm/transformations/PrepareLeave.java
index 0ad37f8bcd..a69ca6f58a 100644
--- a/src/java/org/apache/cassandra/tcm/transformations/PrepareLeave.java
+++ b/src/java/org/apache/cassandra/tcm/transformations/PrepareLeave.java
@@ -121,7 +121,7 @@ public class PrepareLeave implements Transformation
StartLeave start = new StartLeave(leaving, startDelta, unlockKey);
MidLeave mid = new MidLeave(leaving, midDelta, unlockKey);
- FinishLeave leave = new FinishLeave(leaving, finishDelta, unlockKey);
+ FinishLeave leave = new FinishLeave(leaving, finishDelta, unlockKey,
!UnlockSequence.isSupportedBy(prev));
UnbootstrapAndLeave plan =
UnbootstrapAndLeave.newSequence(prev.nextEpoch(),
unlockKey,
@@ -294,7 +294,12 @@ public class PrepareLeave implements Transformation
public FinishLeave(NodeId nodeId, PlacementDeltas delta,
LockedRanges.Key lockKey)
{
- super(nodeId, delta, lockKey, true);
+ this(nodeId, delta, lockKey, true);
+ }
+
+ public FinishLeave(NodeId nodeId, PlacementDeltas delta,
LockedRanges.Key lockKey, boolean unlock)
+ {
+ super(nodeId, delta, lockKey, unlock);
}
@Override
@@ -306,15 +311,48 @@ public class PrepareLeave implements Transformation
@Override
public ClusterMetadata.Transformer transform(ClusterMetadata prev,
ClusterMetadata.Transformer transformer)
{
- return transformer.left(nodeId)
- .with(prev.inProgressSequences.without(nodeId));
+ ClusterMetadata.Transformer next = transformer.left(nodeId);
+ return unlock ? next.with(prev.inProgressSequences.without(nodeId))
+ : next.with(prev.inProgressSequences.with(nodeId,
(UnbootstrapAndLeave plan) -> plan.advance(prev.nextEpoch())));
}
public static class Serializer extends SerializerBase<FinishLeave>
{
+ @Override
+ public void serialize(Transformation t, DataOutputPlus out,
Version version) throws IOException
+ {
+ super.serialize(t, out, version);
+ // TODO (now): confirm that this is the Version to use
+ if (version.isAtLeast(Version.TBD_UNLOCK_STEP))
+ out.writeBoolean(((FinishLeave) t).unlock);
+ }
+
+ @Override
+ public FinishLeave deserialize(DataInputPlus in, Version version)
throws IOException
+ {
+ NodeId nodeId = NodeId.serializer.deserialize(in, version);
+ PlacementDeltas delta =
PlacementDeltas.serializer.deserialize(in, version);
+ LockedRanges.Key lockKey =
LockedRanges.Key.serializer.deserialize(in, version);
+ // UnlockSequence step which follows after FinishLeave was
introduced in TBD_UNLOCK_STEP
+ // to support mutation tracking, and unlock now happens in
that step.
+ // TODO (now): confirm the correct Version to use here.
+ boolean unlock = version.isBefore(Version.TBD_UNLOCK_STEP) ||
in.readBoolean();
+ return new FinishLeave(nodeId, delta, lockKey, unlock);
+ }
+
+ @Override
+ public long serializedSize(Transformation t, Version version)
+ {
+ long size = super.serializedSize(t, version);
+ // TODO (now): confirm the correct Version to use here.
+ if (version.isAtLeast(Version.TBD_UNLOCK_STEP))
+ size += TypeSizes.sizeof(((FinishLeave)t).unlock);
+ return size;
+ }
+
FinishLeave construct(NodeId nodeId, PlacementDeltas delta,
LockedRanges.Key lockKey)
{
- return new FinishLeave(nodeId, delta, lockKey);
+ throw new IllegalStateException();
}
}
}
diff --git a/src/java/org/apache/cassandra/tcm/transformations/PrepareMove.java
b/src/java/org/apache/cassandra/tcm/transformations/PrepareMove.java
index e7e278d0d2..e6b06d9c21 100644
--- a/src/java/org/apache/cassandra/tcm/transformations/PrepareMove.java
+++ b/src/java/org/apache/cassandra/tcm/transformations/PrepareMove.java
@@ -108,7 +108,8 @@ public class PrepareMove implements Transformation
LockedRanges.Key lockKey = LockedRanges.keyFor(prev.nextEpoch());
StartMove startMove = new StartMove(nodeId,
transitionPlan.addToWrites(), lockKey);
MidMove midMove = new MidMove(nodeId, transitionPlan.moveReads(),
lockKey);
- FinishMove finishMove = new FinishMove(nodeId, tokens,
transitionPlan.removeFromWrites(), lockKey);
+ FinishMove finishMove =
+ new FinishMove(nodeId, tokens, transitionPlan.removeFromWrites(),
lockKey, !UnlockSequence.isSupportedBy(prev));
transitionPlan.assertPreExistingWriteReplica(prev.placements);
Move sequence = Move.newSequence(prev.nextEpoch(),
@@ -244,7 +245,12 @@ public class PrepareMove implements Transformation
@VisibleForTesting
public FinishMove(NodeId nodeId, Collection<Token> newTokens,
PlacementDeltas delta, LockedRanges.Key lockKey)
{
- super(nodeId, delta, lockKey, true);
+ this(nodeId, newTokens, delta, lockKey, true);
+ }
+
+ public FinishMove(NodeId nodeId, Collection<Token> newTokens,
PlacementDeltas delta, LockedRanges.Key lockKey, boolean unlock)
+ {
+ super(nodeId, delta, lockKey, unlock);
this.newTokens = newTokens;
}
@@ -257,9 +263,11 @@ public class PrepareMove implements Transformation
@Override
public ClusterMetadata.Transformer transform(ClusterMetadata prev,
ClusterMetadata.Transformer transformer)
{
- return transformer.moveTokens(nodeId, newTokens)
- .join(nodeId)
- .with(prev.inProgressSequences.without(nodeId));
+ ClusterMetadata.Transformer next =
+ transformer.moveTokens(nodeId, newTokens)
+ .join(nodeId);
+ return unlock ? next.with(prev.inProgressSequences.without(nodeId))
+ : next.with(prev.inProgressSequences.with(nodeId,
(Move plan) -> plan.advance(prev.nextEpoch())));
}
public static final class Serializer extends
SerializerBase<PrepareMove.FinishMove>
@@ -272,6 +280,9 @@ public class PrepareMove implements Transformation
out.writeUnsignedVInt32(tokens.size());
for (Token token : tokens)
Token.metadataSerializer.serialize(token, out, version);
+ // TODO (now): confirm that this is the Version to use
+ if (version.isAtLeast(Version.TBD_UNLOCK_STEP))
+ out.writeBoolean(((FinishMove) t).unlock);
}
public FinishMove deserialize(DataInputPlus in, Version version)
throws IOException
@@ -285,7 +296,11 @@ public class PrepareMove implements Transformation
for (int i = 0; i < numTokens; i++)
tokens.add(Token.metadataSerializer.deserialize(in,
partitioner, version));
- return new FinishMove(nodeId, tokens, delta, lockKey);
+ // UnlockSequence step which follows after FinishMove was
introduced in TBD_UNLOCK_STEP
+ // to support mutation tracking, and unlock now happens in
that step.
+ // TODO (now): confirm the correct Version to use here.
+ boolean unlock = version.isBefore(Version.TBD_UNLOCK_STEP) ||
in.readBoolean();
+ return new FinishMove(nodeId, tokens, delta, lockKey, unlock);
}
public long serializedSize(Transformation t, Version version)
@@ -295,6 +310,9 @@ public class PrepareMove implements Transformation
size += TypeSizes.sizeofUnsignedVInt(tokens.size());
for (Token token : tokens)
size += Token.metadataSerializer.serializedSize(token,
version);
+ // TODO (now): confirm the correct Version to use here.
+ if (version.isAtLeast(Version.TBD_UNLOCK_STEP))
+ size += TypeSizes.sizeof(((FinishMove)t).unlock);
return size;
}
diff --git
a/src/java/org/apache/cassandra/tcm/transformations/PrepareReplace.java
b/src/java/org/apache/cassandra/tcm/transformations/PrepareReplace.java
index 8b1577822b..1b45e4c76c 100644
--- a/src/java/org/apache/cassandra/tcm/transformations/PrepareReplace.java
+++ b/src/java/org/apache/cassandra/tcm/transformations/PrepareReplace.java
@@ -108,7 +108,8 @@ public class PrepareReplace implements Transformation
StartReplace start = new StartReplace(replaced, replacement,
transitionPlan.addToWrites(), unlockKey);
MidReplace mid = new MidReplace(replaced, replacement,
transitionPlan.moveReads(), unlockKey);
- FinishReplace finish = new FinishReplace(replaced, replacement,
transitionPlan.removeFromWrites(), unlockKey);
+ FinishReplace finish =
+ new FinishReplace(replaced, replacement,
transitionPlan.removeFromWrites(), unlockKey,
!UnlockSequence.isSupportedBy(prev));
transitionPlan.assertPreExistingWriteReplica(prev.placements);
Set<Token> tokens = new HashSet<>(prev.tokenMap.tokens(replaced));
@@ -327,7 +328,16 @@ public class PrepareReplace implements Transformation
PlacementDeltas delta,
LockedRanges.Key unlockKey)
{
- super(replaced, replacement, delta, unlockKey, true);
+ this(replaced, replacement, delta, unlockKey, true);
+ }
+
+ public FinishReplace(NodeId replaced,
+ NodeId replacement,
+ PlacementDeltas delta,
+ LockedRanges.Key unlockKey,
+ boolean unlock)
+ {
+ super(replaced, replacement, delta, unlockKey, unlock);
}
@Override
@@ -339,8 +349,9 @@ public class PrepareReplace implements Transformation
@Override
public ClusterMetadata.Transformer transform(ClusterMetadata prev,
ClusterMetadata.Transformer transformer)
{
- return transformer.replaced(replaced, replacement())
- .with(prev.inProgressSequences.without(nodeId));
+ ClusterMetadata.Transformer next = transformer.replaced(replaced,
replacement());
+ return unlock ? next.with(prev.inProgressSequences.without(nodeId))
+ : next.with(prev.inProgressSequences.with(nodeId,
(BootstrapAndReplace plan) -> plan.advance(prev.nextEpoch())));
}
@Override
@@ -356,10 +367,43 @@ public class PrepareReplace implements Transformation
public static final class Serializer extends
PrepareReplace.SerializerBase<FinishReplace>
{
+ @Override
+ public void serialize(Transformation t, DataOutputPlus out,
Version version) throws IOException
+ {
+ super.serialize(t, out, version);
+ // TODO (now): confirm that this is the Version to use
+ if (version.isAtLeast(Version.TBD_UNLOCK_STEP))
+ out.writeBoolean(((FinishReplace) t).unlock);
+ }
+
+ @Override
+ public FinishReplace deserialize(DataInputPlus in, Version
version) throws IOException
+ {
+ NodeId replaced = NodeId.serializer.deserialize(in, version);
+ NodeId replacement = NodeId.serializer.deserialize(in,
version);
+ PlacementDeltas delta =
PlacementDeltas.serializer.deserialize(in, version);
+ LockedRanges.Key lockKey =
LockedRanges.Key.serializer.deserialize(in, version);
+ // UnlockSequence step which follows after FinishReplace was
introduced in TBD_UNLOCK_STEP
+ // to support mutation tracking, and unlock now happens in
that step.
+ // TODO (now): confirm the correct Version to use here.
+ boolean unlock = version.isBefore(Version.TBD_UNLOCK_STEP) ||
in.readBoolean();
+ return new FinishReplace(replaced, replacement, delta,
lockKey, unlock);
+ }
+
+ @Override
+ public long serializedSize(Transformation t, Version version)
+ {
+ long size = super.serializedSize(t, version);
+ // TODO (now): confirm the correct Version to use here.
+ if (version.isAtLeast(Version.TBD_UNLOCK_STEP))
+ size += TypeSizes.sizeof(((FinishReplace)t).unlock);
+ return size;
+ }
+
@Override
FinishReplace construct(NodeId replaced, NodeId replacement,
PlacementDeltas delta, LockedRanges.Key lockKey)
{
- return new FinishReplace(replaced, replacement, delta,
lockKey);
+ throw new IllegalStateException();
}
}
}
diff --git
a/src/java/org/apache/cassandra/tcm/transformations/UnlockSequence.java
b/src/java/org/apache/cassandra/tcm/transformations/UnlockSequence.java
new file mode 100644
index 0000000000..088d8fc2ce
--- /dev/null
+++ b/src/java/org/apache/cassandra/tcm/transformations/UnlockSequence.java
@@ -0,0 +1,120 @@
+/*
+ * 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.tcm.transformations;
+
+import java.io.IOException;
+
+import org.apache.cassandra.exceptions.ExceptionCode;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.Transformation;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.tcm.sequences.LockedRanges;
+import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer;
+import org.apache.cassandra.tcm.serialization.Version;
+
+public class UnlockSequence implements Transformation
+{
+ public static final Serializer serializer = new Serializer();
+
+ private final NodeId nodeId;
+ private final LockedRanges.Key lockKey;
+
+ public UnlockSequence(NodeId nodeId, LockedRanges.Key lockKey)
+ {
+ this.nodeId = nodeId;
+ this.lockKey = lockKey;
+ }
+
+ static boolean isSupportedBy(ClusterMetadata metadata)
+ {
+ return
metadata.directory.commonSerializationVersion.isAtLeast(Kind.UNLOCK_SEQUENCE.introducedIn);
+ }
+
+ @Override
+ public Kind kind()
+ {
+ return Kind.UNLOCK_SEQUENCE;
+ }
+
+ @Override
+ public Result execute(ClusterMetadata prev)
+ {
+ if (!prev.inProgressSequences.contains(nodeId))
+ return new Rejected(ExceptionCode.INVALID, "Can't find an
in-progress sequence for this operation");
+
+ if (prev.inProgressSequences.get(nodeId).nextStep() != kind())
+ return new Rejected(ExceptionCode.INVALID, String.format("Can't
commit sequenced operations out of order. Expected %s, but got %s",
prev.inProgressSequences.get(nodeId).nextStep(), kind()));
+
+ ClusterMetadata.Transformer next =
+ prev.transformer()
+ .with(prev.inProgressSequences.without(nodeId))
+ .with(prev.lockedRanges.unlock(lockKey));
+
+ return Transformation.success(next,
prev.lockedRanges.locked.get(lockKey));
+ }
+
+ @Override
+ public String toString()
+ {
+ return getClass().getSimpleName() + '{' + "id=" + nodeId + '}';
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) return true;
+ if (!(o instanceof UnlockSequence)) return false;
+ UnlockSequence that = (UnlockSequence) o;
+ return this.nodeId.equals(that.nodeId) &&
this.lockKey.equals(that.lockKey);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return nodeId.hashCode() + 31 * lockKey.hashCode();
+ }
+
+ public static final class Serializer implements
AsymmetricMetadataSerializer<Transformation, UnlockSequence>
+ {
+ @Override
+ public void serialize(Transformation t, DataOutputPlus out, Version
version) throws IOException
+ {
+ UnlockSequence unlock = (UnlockSequence) t;
+ NodeId.serializer.serialize(unlock.nodeId, out, version);
+ LockedRanges.Key.serializer.serialize(unlock.lockKey, out,
version);
+ }
+
+ @Override
+ public UnlockSequence deserialize(DataInputPlus in, Version version)
throws IOException
+ {
+ NodeId nodeId = NodeId.serializer.deserialize(in, version);
+ LockedRanges.Key lockKey =
LockedRanges.Key.serializer.deserialize(in, version);
+ return new UnlockSequence(nodeId, lockKey);
+ }
+
+ @Override
+ public long serializedSize(Transformation t, Version version)
+ {
+ UnlockSequence unlock = (UnlockSequence) t;
+ return NodeId.serializer.serializedSize(unlock.nodeId, version) +
+ LockedRanges.Key.serializer.serializedSize(unlock.lockKey,
version);
+ }
+ }
+}
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/MoveTest.java
b/test/distributed/org/apache/cassandra/distributed/test/MoveTest.java
index a5517dc043..981c168ff0 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/MoveTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/MoveTest.java
@@ -50,7 +50,7 @@ public class MoveTest extends TestBaseImpl
private void move(boolean forwards) throws Throwable
{
try (Cluster cluster = Cluster.build(4)
- .withConfig(config ->
config.set("paxos_variant",
"v2_without_linearizable_reads").with(NETWORK).with(GOSSIP))
+ .withConfig(config ->
config.set("paxos_variant",
"v2_without_linearizable_reads").set("mutation_tracking.enabled",
false).with(NETWORK).with(GOSSIP))
.withoutVNodes()
.start())
{
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java
b/test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java
index 4492d1e15d..ae1f8c21b7 100644
---
a/test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java
+++
b/test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java
@@ -69,6 +69,7 @@ import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Commit;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.MetadataSnapshots;
+import org.apache.cassandra.tcm.MultiStepOperation;
import org.apache.cassandra.tcm.RegistrationStatus;
import org.apache.cassandra.tcm.Transformation;
import org.apache.cassandra.tcm.log.LocalLog;
@@ -97,6 +98,7 @@ import org.apache.cassandra.tcm.transformations.PrepareMove;
import org.apache.cassandra.tcm.transformations.PrepareReplace;
import org.apache.cassandra.tcm.transformations.Register;
import org.apache.cassandra.tcm.transformations.Unregister;
+import org.apache.cassandra.tcm.transformations.UnlockSequence;
import org.apache.cassandra.tcm.transformations.cms.AdvanceCMSReconfiguration;
import org.apache.cassandra.tcm.transformations.cms.PrepareCMSReconfiguration;
import org.apache.cassandra.utils.ByteBufferUtil;
@@ -533,6 +535,7 @@ public class ClusterMetadataTestHelper
BootstrapAndJoin plan = getBootstrapPlan(endpoint);
assert plan.next == Transformation.Kind.FINISH_JOIN;
commit(plan.finishJoin);
+ maybeCommitUnlock(plan.startJoin.nodeId(), plan.lockKey);
idx++;
return this;
}
@@ -632,6 +635,7 @@ public class ClusterMetadataTestHelper
UnbootstrapAndLeave plan = getLeavePlan(endpoint);
assert plan.next == Transformation.Kind.FINISH_LEAVE;
commit(plan.finishLeave);
+ maybeCommitUnlock(plan.startLeave.nodeId(), plan.lockKey);
idx++;
return this;
}
@@ -741,6 +745,7 @@ public class ClusterMetadataTestHelper
BootstrapAndReplace plan = getReplacePlan(replacement);
assert plan.next == Transformation.Kind.FINISH_REPLACE;
commit(plan.finishReplace);
+ maybeCommitUnlock(plan.startReplace.nodeId(),
plan.lockKey);
idx++;
return this;
}
@@ -819,6 +824,7 @@ public class ClusterMetadataTestHelper
Move plan = getMovePlan(endpoint);
assert plan.next == Transformation.Kind.FINISH_MOVE;
commit(plan.finishMove);
+ maybeCommitUnlock(plan.startMove.nodeId(), plan.lockKey);
idx++;
return this;
}
@@ -944,6 +950,16 @@ public class ClusterMetadataTestHelper
return ClusterMetadataService.instance().commit(transform);
}
+ // On UNLOCK_SEQUENCE-supporting clusters the FINISH step no longer
unlocks/retires the sequence; a trailing
+ // UNLOCK_SEQUENCE step does. Commit it here so the step-driving helpers
observe a retired sequence, matching
+ // the legacy (pre-UNLOCK_SEQUENCE) behaviour where FINISH did both.
+ private static void maybeCommitUnlock(NodeId nodeId, LockedRanges.Key
lockKey) throws ExecutionException, InterruptedException
+ {
+ MultiStepOperation<?> seq =
ClusterMetadata.current().inProgressSequences.get(nodeId);
+ if (seq != null && seq.nextStep() ==
Transformation.Kind.UNLOCK_SEQUENCE)
+ commit(new UnlockSequence(nodeId, lockKey));
+ }
+
public static interface NodeOperations
{
public void register();
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/log/MetadataKeysTest.java
b/test/distributed/org/apache/cassandra/distributed/test/log/MetadataKeysTest.java
index acc013aff0..a0adf6c702 100644
---
a/test/distributed/org/apache/cassandra/distributed/test/log/MetadataKeysTest.java
+++
b/test/distributed/org/apache/cassandra/distributed/test/log/MetadataKeysTest.java
@@ -40,6 +40,7 @@ import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.MetadataKey;
import org.apache.cassandra.tcm.MetadataKeys;
+import org.apache.cassandra.tcm.MultiStepOperation;
import org.apache.cassandra.tcm.Transformation;
import org.apache.cassandra.tcm.membership.Location;
import org.apache.cassandra.tcm.membership.NodeAddresses;
@@ -48,6 +49,7 @@ import org.apache.cassandra.tcm.membership.NodeState;
import org.apache.cassandra.tcm.membership.NodeVersion;
import org.apache.cassandra.tcm.ownership.PlacementProvider;
import org.apache.cassandra.tcm.sequences.BootstrapAndJoin;
+import org.apache.cassandra.tcm.sequences.LockedRanges;
import org.apache.cassandra.tcm.sequences.LeaveStreams;
import org.apache.cassandra.tcm.sequences.Move;
import org.apache.cassandra.tcm.sequences.UnbootstrapAndLeave;
@@ -56,6 +58,7 @@ import org.apache.cassandra.tcm.transformations.PrepareJoin;
import org.apache.cassandra.tcm.transformations.PrepareLeave;
import org.apache.cassandra.tcm.transformations.PrepareMove;
import org.apache.cassandra.tcm.transformations.Register;
+import org.apache.cassandra.tcm.transformations.UnlockSequence;
public class MetadataKeysTest extends CMSTestBase
{
@@ -97,6 +100,7 @@ public class MetadataKeysTest extends CMSTestBase
checkDiff(sut.service, bootstrapAndJoin.finishJoin);
sut.service.commit(bootstrapAndJoin.finishJoin);
+ maybeCommitUnlock(sut.service, prepareJoin.nodeId(),
bootstrapAndJoin.lockKey);
}
{
@@ -113,6 +117,7 @@ public class MetadataKeysTest extends CMSTestBase
checkDiff(sut.service, bootstrapAndMove.finishMove);
sut.service.commit(bootstrapAndMove.finishMove);
+ maybeCommitUnlock(sut.service, prepareMove.nodeId(),
bootstrapAndMove.lockKey);
}
{
@@ -129,6 +134,7 @@ public class MetadataKeysTest extends CMSTestBase
checkDiff(sut.service, bootstrapAndLeave.finishLeave);
sut.service.commit(bootstrapAndLeave.finishLeave);
+ maybeCommitUnlock(sut.service, prepareLeave.nodeId(),
bootstrapAndLeave.lockKey);
}
}
}
@@ -153,6 +159,17 @@ public class MetadataKeysTest extends CMSTestBase
}
}
+ private static void maybeCommitUnlock(ClusterMetadataService cms, NodeId
nodeId, LockedRanges.Key lockKey)
+ {
+ MultiStepOperation<?> seq =
cms.metadata().inProgressSequences.get(nodeId);
+ if (seq != null && seq.nextStep() ==
Transformation.Kind.UNLOCK_SEQUENCE)
+ {
+ UnlockSequence unlock = new UnlockSequence(nodeId, lockKey);
+ checkDiff(cms, unlock);
+ cms.commit(unlock);
+ }
+ }
+
private static void checkDiff(ClusterMetadataService cms, Transformation
transformation)
{
ClusterMetadata before = cms.metadata();
diff --git
a/test/simulator/main/org/apache/cassandra/simulator/cluster/OnClusterJoin.java
b/test/simulator/main/org/apache/cassandra/simulator/cluster/OnClusterJoin.java
index bb36ad7218..58a651c53d 100644
---
a/test/simulator/main/org/apache/cassandra/simulator/cluster/OnClusterJoin.java
+++
b/test/simulator/main/org/apache/cassandra/simulator/cluster/OnClusterJoin.java
@@ -113,6 +113,13 @@ class OnClusterJoin extends OnClusterChangeTopology
assert bootstrapAndJoin.next.ordinal() == kind :
String.format("Expected next step to be %s, but got %s",
Transformation.Kind.values()[kind], bootstrapAndJoin.next);
boolean res = bootstrapAndJoin.executeNext().isContinuable();
assert res;
+
+ // On UNLOCK_SEQUENCE-supporting clusters FINISH_JOIN no
longer unlocks/retires the sequence;
+ // drain the trailing UNLOCK_SEQUENCE step so the range lock
is released.
+ ClusterMetadata after = ClusterMetadata.current();
+ MultiStepOperation<?> trailing =
after.inProgressSequences.get(after.myNodeId());
+ if (trailing != null && trailing.nextStep() ==
Transformation.Kind.UNLOCK_SEQUENCE)
+ assert trailing.executeNext().isContinuable();
});
}
}
diff --git
a/test/simulator/main/org/apache/cassandra/simulator/cluster/OnClusterLeave.java
b/test/simulator/main/org/apache/cassandra/simulator/cluster/OnClusterLeave.java
index 2b2ce15086..7cd9fe7a6d 100644
---
a/test/simulator/main/org/apache/cassandra/simulator/cluster/OnClusterLeave.java
+++
b/test/simulator/main/org/apache/cassandra/simulator/cluster/OnClusterLeave.java
@@ -116,6 +116,13 @@ class OnClusterLeave extends OnClusterChangeTopology
assert unbootstrapAndLeave.next.ordinal() == kind :
String.format("Expected next step to be %s, but got %s",
Transformation.Kind.values()[kind], unbootstrapAndLeave.next);
boolean res =
unbootstrapAndLeave.executeNext().isContinuable();
assert res;
+
+ // On UNLOCK_SEQUENCE-supporting clusters FINISH_LEAVE no
longer unlocks/retires the sequence;
+ // drain the trailing UNLOCK_SEQUENCE step so the range lock
is released.
+ ClusterMetadata after = ClusterMetadata.current();
+ MultiStepOperation<?> trailing =
after.inProgressSequences.get(metadata.myNodeId());
+ if (trailing != null && trailing.nextStep() ==
Transformation.Kind.UNLOCK_SEQUENCE)
+ assert trailing.executeNext().isContinuable();
});
}
}
diff --git
a/test/simulator/main/org/apache/cassandra/simulator/cluster/OnClusterReplace.java
b/test/simulator/main/org/apache/cassandra/simulator/cluster/OnClusterReplace.java
index 140da9f18e..e7e6a03d45 100644
---
a/test/simulator/main/org/apache/cassandra/simulator/cluster/OnClusterReplace.java
+++
b/test/simulator/main/org/apache/cassandra/simulator/cluster/OnClusterReplace.java
@@ -168,6 +168,13 @@ class OnClusterReplace extends OnClusterChangeTopology
assert bootstrapAndReplace.next.ordinal() == kind :
String.format("Expected next step to be %s, but got %s",
Transformation.Kind.values()[kind], bootstrapAndReplace.next);
boolean res =
bootstrapAndReplace.executeNext().isContinuable();
assert res;
+
+ // On UNLOCK_SEQUENCE-supporting clusters FINISH_REPLACE no
longer unlocks/retires the sequence;
+ // drain the trailing UNLOCK_SEQUENCE step so the range lock
is released.
+ ClusterMetadata after = ClusterMetadata.current();
+ MultiStepOperation<?> trailing =
after.inProgressSequences.get(metadata.myNodeId());
+ if (trailing != null && trailing.nextStep() ==
Transformation.Kind.UNLOCK_SEQUENCE)
+ assert trailing.executeNext().isContinuable();
});
}
}
diff --git
a/test/unit/org/apache/cassandra/tcm/transformations/EventsMetadataTest.java
b/test/unit/org/apache/cassandra/tcm/transformations/EventsMetadataTest.java
index 11fd03c696..bb767c94c1 100644
--- a/test/unit/org/apache/cassandra/tcm/transformations/EventsMetadataTest.java
+++ b/test/unit/org/apache/cassandra/tcm/transformations/EventsMetadataTest.java
@@ -136,6 +136,7 @@ public class EventsMetadataTest
ClusterMetadataService.instance().commit(plan.midJoin);
ClusterMetadataService.instance().commit(plan.finishJoin);
+ ClusterMetadataService.instance().commit(new
org.apache.cassandra.tcm.transformations.UnlockSequence(nodeId, plan.lockKey));
assertEquals(NodeState.JOINED,
ClusterMetadata.current().directory.peerState(nodeId));
assertTrue(ClusterMetadata.current().lockedRanges.locked.isEmpty());
@@ -170,6 +171,7 @@ public class EventsMetadataTest
ClusterMetadataService.instance().commit(join.startJoin);
ClusterMetadataService.instance().commit(join.midJoin);
ClusterMetadataService.instance().commit(join.finishJoin);
+ ClusterMetadataService.instance().commit(new
org.apache.cassandra.tcm.transformations.UnlockSequence(nodeId, join.lockKey));
ClusterMetadata before = ClusterMetadata.current();
ClusterMetadataService.instance().commit(new PrepareLeave(nodeId,
true, PrepareLeaveTest.dummyPlacementProvider, LeaveStreams.Kind.UNBOOTSTRAP));;
@@ -184,6 +186,7 @@ public class EventsMetadataTest
ClusterMetadataService.instance().commit(leave.startLeave);
ClusterMetadataService.instance().commit(leave.midLeave);
ClusterMetadataService.instance().commit(leave.finishLeave);
+ ClusterMetadataService.instance().commit(new
org.apache.cassandra.tcm.transformations.UnlockSequence(nodeId, leave.lockKey));
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
