From 9d856f2c08374653a50b646a6cb46d63ca2ae4ac Mon Sep 17 00:00:00 2001 From: Aleksey Yeschenko Date: Thu, 17 Jul 2025 13:00:33 +0100 Subject: [PATCH] Implement Shard and CoordinatorLog metadata durability patch by Aleksey Yeschenko; reviewed by Abe Ratnofsky for CASSANDRA-20882 --- .../apache/cassandra/db/SystemKeyspace.java | 80 ++++- .../io/sstable/format/SSTableWriter.java | 2 +- .../replication/BroadcastLogOffsets.java | 12 +- .../cassandra/replication/CoordinatorLog.java | 308 ++++++++++++++---- .../replication/MutationTrackingService.java | 237 ++++++++------ .../replication/Node2OffsetsMap.java | 89 ++++- .../apache/cassandra/replication/Offsets.java | 33 ++ .../cassandra/replication/Participants.java | 20 ++ .../apache/cassandra/replication/Shard.java | 188 +++++++---- .../replication/UnreconciledMutations.java | 38 +++ .../tools/OfflineClusterMetadataDump.java | 4 +- .../test/log/SystemKeyspaceStorageTest.java | 4 +- .../cql3/SystemKeyspaceQueryTest.java | 4 +- .../CoordinatorLogOffsetsLifecycleTest.java | 19 +- .../CoordinatorLogOffsetsTest.java | 43 ++- .../replication/CoordinatorLogTest.java | 123 ++++++- .../cassandra/replication/OffsetsTest.java | 12 +- .../cassandra/replication/ShardTest.java | 112 +++++++ .../UnreconciledMutationsTest.java | 115 ++++--- 19 files changed, 1102 insertions(+), 341 deletions(-) create mode 100644 test/unit/org/apache/cassandra/replication/ShardTest.java diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java index d4e4912264..23438dbf83 100644 --- a/src/java/org/apache/cassandra/db/SystemKeyspace.java +++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java @@ -190,7 +190,10 @@ public final class SystemKeyspace public static final String REPAIRS = "repairs"; public static final String TOP_PARTITIONS = "top_partitions"; public static final String METADATA_LOG = "local_metadata_log"; - public static final String SNAPSHOT_TABLE_NAME = "metadata_snapshots"; + public static final String METADATA_SNAPSHOTS = "metadata_snapshots"; + public static final String HOST_LOG_ID = "host_log_id"; + public static final String SHARDS = "shards"; + public static final String COORDINATOR_LOGS = "coordinator_logs"; /** * By default the system keyspace tables should be stored in a single data directory to allow the server @@ -225,14 +228,16 @@ public final class SystemKeyspace TABLE_ESTIMATES_TYPE_LOCAL_PRIMARY, AVAILABLE_RANGES_V2, TRANSFERRED_RANGES_V2, VIEW_BUILDS_IN_PROGRESS, BUILT_VIEWS, PREPARED_STATEMENTS, REPAIRS, TOP_PARTITIONS, LEGACY_PEERS, LEGACY_PEER_EVENTS, LEGACY_TRANSFERRED_RANGES, LEGACY_AVAILABLE_RANGES, LEGACY_SIZE_ESTIMATES, LEGACY_SSTABLE_ACTIVITY, - METADATA_LOG, SNAPSHOT_TABLE_NAME, CONSENSUS_MIGRATION_STATE); + METADATA_LOG, METADATA_SNAPSHOTS, CONSENSUS_MIGRATION_STATE, + HOST_LOG_ID, SHARDS, COORDINATOR_LOGS); public static final Set TABLE_NAMES = ImmutableSet.of( BATCHES, PAXOS, PAXOS_REPAIR_HISTORY, BUILT_INDEXES, LOCAL, PEERS_V2, PEER_EVENTS_V2, COMPACTION_HISTORY, SSTABLE_ACTIVITY_V2, TABLE_ESTIMATES, AVAILABLE_RANGES_V2, TRANSFERRED_RANGES_V2, VIEW_BUILDS_IN_PROGRESS, BUILT_VIEWS, PREPARED_STATEMENTS, REPAIRS, TOP_PARTITIONS, LEGACY_PEERS, LEGACY_PEER_EVENTS, LEGACY_TRANSFERRED_RANGES, LEGACY_AVAILABLE_RANGES, LEGACY_SIZE_ESTIMATES, LEGACY_SSTABLE_ACTIVITY, - METADATA_LOG, SNAPSHOT_TABLE_NAME, CONSENSUS_MIGRATION_STATE); + METADATA_LOG, METADATA_SNAPSHOTS, CONSENSUS_MIGRATION_STATE, + HOST_LOG_ID, SHARDS, COORDINATOR_LOGS); public static final TableMetadata Batches = parse(BATCHES, @@ -518,7 +523,7 @@ public final class SystemKeyspace + "cfids set, " + "PRIMARY KEY (parent_id))").build(); - public static final TableMetadata LocalMetadataLog = + private static final TableMetadata LocalMetadataLog = parse(METADATA_LOG, "Local Metadata Log", "CREATE TABLE %s (" @@ -532,13 +537,49 @@ public final class SystemKeyspace "compaction_window_size","1"))) .build(); - public static final TableMetadata Snapshots = parse(SNAPSHOT_TABLE_NAME, - "ClusterMetadata snapshots", - "CREATE TABLE IF NOT EXISTS %s (" + - "epoch bigint PRIMARY KEY," + - "snapshot blob)") - .partitioner(MetaStrategy.partitioner) - .build(); + private static final TableMetadata Snapshots = + parse(METADATA_SNAPSHOTS, + "ClusterMetadata snapshots", + "CREATE TABLE IF NOT EXISTS %s (" + + "epoch bigint PRIMARY KEY," + + "snapshot blob)") + .partitioner(MetaStrategy.partitioner) + .build(); + + private static final TableMetadata HostLogId = + parse(HOST_LOG_ID, + "mutation tracking last used host log id", + "CREATE TABLE %s (" + + "key text," + + "host_log_id int," + + "PRIMARY KEY ((key)))") + .build(); + + private static final TableMetadata Shards = + parse(SHARDS, + "mutation tracking shards", + "CREATE TABLE %s (" + + "keyspace_name text," + + "range_start text," + + "range_end text," + + "participants frozen>," + + "PRIMARY KEY ((keyspace_name, range_start, range_end)))") + .build(); + + private static final TableMetadata CoordinatorLogs = + parse(COORDINATOR_LOGS, + "mutation tracking coordinator logs", + "CREATE TABLE %s (" + + "keyspace_name text," + + "range_start text," + + "range_end text," + + "host_id int," + + "host_log_id int," + + "participants frozen>," + + "witnessed_offsets map>>," + + "persisted_offsets map>>," + + "PRIMARY KEY ((keyspace_name, range_start, range_end), host_id, host_log_id))") + .build(); @Deprecated(since = "4.0") private static final TableMetadata LegacyPeers = @@ -633,7 +674,10 @@ public final class SystemKeyspace TopPartitions, LocalMetadataLog, Snapshots, - ConsensusMigrationState); + ConsensusMigrationState, + HostLogId, + Shards, + CoordinatorLogs); } private static volatile Map> truncationRecords; @@ -2129,16 +2173,16 @@ public final class SystemKeyspace { Preconditions.checkArgument(epoch.isAfter(Epoch.FIRST), "Cannot store a snapshot for an epoch less than " + Epoch.FIRST.getEpoch()); logger.info("Storing snapshot of cluster metadata at epoch {}", epoch); - String query = String.format("INSERT INTO %s.%s (epoch, snapshot) VALUES (?, ?)", SchemaConstants.SYSTEM_KEYSPACE_NAME, SNAPSHOT_TABLE_NAME); + String query = String.format("INSERT INTO %s.%s (epoch, snapshot) VALUES (?, ?)", SchemaConstants.SYSTEM_KEYSPACE_NAME, METADATA_SNAPSHOTS); executeInternal(query, epoch.getEpoch(), snapshot); - forceBlockingFlush(SNAPSHOT_TABLE_NAME); + forceBlockingFlush(METADATA_SNAPSHOTS); } public static ByteBuffer getSnapshot(Epoch epoch) { Preconditions.checkArgument(epoch.isAfter(Epoch.FIRST), "Cannot retrieve a snapshot for an epoch less than " + Epoch.FIRST.getEpoch()); logger.info("Getting snapshot of epoch = {}", epoch); - String query = String.format("SELECT snapshot FROM %s.%s WHERE epoch = ?", SchemaConstants.SYSTEM_KEYSPACE_NAME, SNAPSHOT_TABLE_NAME); + String query = String.format("SELECT snapshot FROM %s.%s WHERE epoch = ?", SchemaConstants.SYSTEM_KEYSPACE_NAME, METADATA_SNAPSHOTS); UntypedResultSet res = executeInternal(query, epoch.getEpoch()); if (res == null || res.isEmpty()) return null; @@ -2167,7 +2211,7 @@ public final class SystemKeyspace { // during gossip upgrade we have epoch = Long.MIN_VALUE + 1 (and the reverse partitioner doesn't support negative keys) search = search.isBefore(Epoch.EMPTY) ? Epoch.EMPTY : search; - String query = String.format("SELECT snapshot FROM %s.%s WHERE token(epoch) >= token(?) LIMIT 1", SchemaConstants.SYSTEM_KEYSPACE_NAME, SNAPSHOT_TABLE_NAME); + String query = String.format("SELECT snapshot FROM %s.%s WHERE token(epoch) >= token(?) LIMIT 1", SchemaConstants.SYSTEM_KEYSPACE_NAME, METADATA_SNAPSHOTS); UntypedResultSet res = executeInternal(query, search.getEpoch()); if (res != null && !res.isEmpty()) return res.one().getBytes("snapshot").duplicate(); @@ -2178,7 +2222,7 @@ public final class SystemKeyspace { // during gossip upgrade we have epoch = Long.MIN_VALUE + 1 (and the reverse partitioner doesn't support negative keys) search = search.isBefore(Epoch.EMPTY) ? Epoch.EMPTY : search; - String query = String.format("SELECT epoch FROM %s.%s WHERE token(epoch) < token(?)", SchemaConstants.SYSTEM_KEYSPACE_NAME, SNAPSHOT_TABLE_NAME); + String query = String.format("SELECT epoch FROM %s.%s WHERE token(epoch) < token(?)", SchemaConstants.SYSTEM_KEYSPACE_NAME, METADATA_SNAPSHOTS); UntypedResultSet res = executeInternal(query, search.getEpoch()); if (res == null) return Collections.emptyList(); @@ -2191,7 +2235,7 @@ public final class SystemKeyspace */ public static ByteBuffer findLastSnapshot() { - String query = String.format("SELECT snapshot FROM %s.%s LIMIT 1", SchemaConstants.SYSTEM_KEYSPACE_NAME, SNAPSHOT_TABLE_NAME); + String query = String.format("SELECT snapshot FROM %s.%s LIMIT 1", SchemaConstants.SYSTEM_KEYSPACE_NAME, METADATA_SNAPSHOTS); UntypedResultSet res = executeInternal(query); if (res != null && !res.isEmpty()) return res.one().getBytes("snapshot").duplicate(); diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java index bef49b94c1..bd466595af 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java @@ -357,7 +357,7 @@ public abstract class SSTableWriter extends SSTable implements Transactional if (metadata().replicationType().isTracked() && repairedAt == ActiveRepairService.UNREPAIRED_SSTABLE) { Preconditions.checkState(Objects.equals(pendingRepair, ActiveRepairService.NO_PENDING_REPAIR)); - if (MutationTrackingService.instance.isDurablyReconciled(getKeyspaceName(), coordinatorLogOffsets)) + if (MutationTrackingService.instance.isDurablyReconciled(coordinatorLogOffsets)) { repairedAt = Clock.Global.currentTimeMillis(); logger.debug("Marking SSTable {} as reconciled with repairedAt {}", descriptor, repairedAt); diff --git a/src/java/org/apache/cassandra/replication/BroadcastLogOffsets.java b/src/java/org/apache/cassandra/replication/BroadcastLogOffsets.java index b3fc118613..ce8fe512c9 100644 --- a/src/java/org/apache/cassandra/replication/BroadcastLogOffsets.java +++ b/src/java/org/apache/cassandra/replication/BroadcastLogOffsets.java @@ -41,12 +41,14 @@ public class BroadcastLogOffsets private final String keyspace; private final Range range; private final List replicatedOffsets; + private final boolean durable; - public BroadcastLogOffsets(String keyspace, Range range, List offsets) + public BroadcastLogOffsets(String keyspace, Range range, List offsets, boolean durable) { this.keyspace = keyspace; this.range = range; this.replicatedOffsets = offsets; + this.durable = durable; } boolean isEmpty() @@ -57,7 +59,7 @@ public class BroadcastLogOffsets @Override public String toString() { - return "ShardReplicatedOffsets{" + keyspace + ", " + range + ", " + replicatedOffsets + '}'; + return "ShardReplicatedOffsets{" + keyspace + ", " + range + ", " + replicatedOffsets + ", " + durable + '}'; } public static final IVerbHandler verbHandler = message -> { @@ -66,6 +68,7 @@ public class BroadcastLogOffsets MutationTrackingService.instance.updateReplicatedOffsets(replicatedOffsets.keyspace, replicatedOffsets.range, replicatedOffsets.replicatedOffsets, + replicatedOffsets.durable, message.from()); }; @@ -79,6 +82,7 @@ public class BroadcastLogOffsets out.writeInt(status.replicatedOffsets.size()); for (Offsets.Immutable logOffsets : status.replicatedOffsets) Offsets.serializer.serialize(logOffsets, out, version); + out.writeBoolean(status.durable); } @Override @@ -90,7 +94,8 @@ public class BroadcastLogOffsets List replicatedOffsets = new ArrayList<>(count); for (int i = 0; i < count; ++i) replicatedOffsets.add(Offsets.serializer.deserialize(in, version)); - return new BroadcastLogOffsets(keyspace, range, replicatedOffsets); + boolean durable = in.readBoolean(); + return new BroadcastLogOffsets(keyspace, range, replicatedOffsets, durable); } @Override @@ -102,6 +107,7 @@ public class BroadcastLogOffsets size += TypeSizes.sizeof(replicatedOffsets.replicatedOffsets.size()); for (Offsets.Immutable logOffsets : replicatedOffsets.replicatedOffsets) size += Offsets.serializer.serializedSize(logOffsets, version); + size += TypeSizes.sizeof(replicatedOffsets.durable); return size; } }; diff --git a/src/java/org/apache/cassandra/replication/CoordinatorLog.java b/src/java/org/apache/cassandra/replication/CoordinatorLog.java index 13228256e2..312c355237 100644 --- a/src/java/org/apache/cassandra/replication/CoordinatorLog.java +++ b/src/java/org/apache/cassandra/replication/CoordinatorLog.java @@ -17,7 +17,11 @@ */ package org.apache.cassandra.replication; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; import java.util.Objects; +import java.util.Set; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; @@ -27,74 +31,101 @@ import javax.annotation.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; +import org.agrona.collections.Int2ObjectHashMap; import org.agrona.collections.IntArrayList; +import org.apache.cassandra.cql3.UntypedResultSet; import org.apache.cassandra.db.Mutation; import org.apache.cassandra.db.PartitionPosition; +import org.apache.cassandra.db.SystemKeyspace; +import org.apache.cassandra.db.marshal.Int32Type; +import org.apache.cassandra.db.marshal.ListType; import org.apache.cassandra.dht.AbstractBounds; +import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; +import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.tcm.ClusterMetadata; +import static java.lang.String.format; +import static org.apache.cassandra.cql3.QueryProcessor.executeInternal; +import static org.apache.cassandra.replication.Node2OffsetsMap.forParticipants; +import static org.apache.cassandra.replication.Node2OffsetsMap.fromPrimitiveMap; import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis; public abstract class CoordinatorLog { private static final Logger logger = LoggerFactory.getLogger(CoordinatorLog.class); - protected final int localHostId; + protected final int localNodeId; + protected final String keyspace; + protected final Range range; protected final CoordinatorLogId logId; protected final Participants participants; + protected final Node2OffsetsMap witnessedOffsets; + protected final Node2OffsetsMap persistedOffsets; + protected final UnreconciledMutations unreconciledMutations; - protected final Offsets.Mutable[] witnessedOffsets; protected final Offsets.Mutable reconciledOffsets; + protected final Offsets.Mutable reconciledPersistedOffsets; protected final ReadWriteLock lock; - abstract void receivedWriteResponse(ShortMutationId mutationId, int fromHostId); + abstract void receivedWriteResponse(ShortMutationId mutationId, int fromNodeId); - CoordinatorLog(int localHostId, CoordinatorLogId logId, Participants participants) + CoordinatorLog(String keyspace, + Range range, + int localNodeId, + CoordinatorLogId logId, + Participants participants, + Node2OffsetsMap witnessedOffsets, + Node2OffsetsMap persistedOffsets, + UnreconciledMutations unreconciledMutations) { - this.localHostId = localHostId; + this.localNodeId = localNodeId; + this.keyspace = keyspace; + this.range = range; this.logId = logId; this.participants = participants; - - this.unreconciledMutations = new UnreconciledMutations(); + this.unreconciledMutations = unreconciledMutations; + this.witnessedOffsets = witnessedOffsets; + this.reconciledOffsets = witnessedOffsets.intersection(); + this.persistedOffsets = persistedOffsets; + this.reconciledPersistedOffsets = persistedOffsets.intersection(); this.lock = new ReentrantReadWriteLock(); - - Offsets.Mutable[] ids = new Offsets.Mutable[participants.size()]; - for (int i = 0; i < participants.size(); i++) - ids[i] = new Offsets.Mutable(logId); - - witnessedOffsets = ids; - reconciledOffsets = new Offsets.Mutable(logId); } - static CoordinatorLog create(int localHostId, CoordinatorLogId id, Participants participants) + CoordinatorLog(String keyspace, Range range, int localNodeId, CoordinatorLogId logId, Participants participants) { - return id.hostId == localHostId ? new CoordinatorLogPrimary(localHostId, id, participants) - : new CoordinatorLogReplica(localHostId, id, participants); + this(keyspace, range, localNodeId, logId, participants, forParticipants(logId, participants), forParticipants(logId, participants), new UnreconciledMutations()); } - void updateReplicatedOffsets(Offsets offsets, int onHostId) + static CoordinatorLog create(String keyspace, Range range, int localNodeId, CoordinatorLogId id, Participants participants) + { + return id.hostId == localNodeId ? new CoordinatorLogPrimary(keyspace, range, localNodeId, id, participants) + : new CoordinatorLogReplica(keyspace, range, localNodeId, id, participants); + } + + static CoordinatorLog recreate( + String keyspace, Range range, int localNodeId, CoordinatorLogId id, Participants participants, + Node2OffsetsMap witnessedOffsets, Node2OffsetsMap persistedOffsets, UnreconciledMutations unreconciledMutations) + { + return id.hostId == localNodeId ? new CoordinatorLogPrimary(keyspace, range, localNodeId, id, participants, witnessedOffsets, persistedOffsets, unreconciledMutations) + : new CoordinatorLogReplica(keyspace, range, localNodeId, id, participants, witnessedOffsets, persistedOffsets, unreconciledMutations); + } + + void updateReplicatedOffsets(Offsets offsets, boolean persisted, int onNodeId) { lock.writeLock().lock(); try { - get(onHostId).addAll(offsets, (ignore, start, end) -> - { - for (int offset = start; offset <= end; ++offset) - { - // TODO (desired): use the fact that Offsets are ordered to optimise this look up - if (othersWitnessed(offset, onHostId)) - { - reconciledOffsets.add(offset); - unreconciledMutations.remove(offset); - } - } - }); + if (persisted) + updatePersistedReplicatedOffsets(offsets, onNodeId); + else + updateWitnessedReplicatedOffsets(offsets, onNodeId); } finally { @@ -102,13 +133,37 @@ public abstract class CoordinatorLog } } + private void updateWitnessedReplicatedOffsets(Offsets offsets, int onNodeId) + { + witnessedOffsets.get(onNodeId).addAll(offsets, (ignore, start, end) -> + { + for (int offset = start; offset <= end; ++offset) + { + // TODO (desired): use the fact that Offsets are ordered to optimise this look up + if (othersWitnessed(offset, onNodeId)) + { + reconciledOffsets.add(offset); + unreconciledMutations.remove(offset); + } + } + }); + } + + private void updatePersistedReplicatedOffsets(Offsets offsets, int onNodeId) + { + persistedOffsets.get(onNodeId).addAll(offsets); + reconciledPersistedOffsets.addAll(persistedOffsets.intersection()); + } + @Nullable - Offsets.Immutable collectReplicatedOffsets() + Offsets.Immutable collectReplicatedOffsets(boolean persisted) { lock.readLock().lock(); try { - Offsets offsets = witnessedOffsets[participants.indexOf(localHostId)]; + Offsets offsets = persisted + ? persistedOffsets.get(localNodeId) + : witnessedOffsets.get(localNodeId); return offsets.isEmpty() ? null : Offsets.Immutable.copy(offsets); } finally @@ -122,7 +177,7 @@ public abstract class CoordinatorLog lock.writeLock().lock(); try { - if (getLocal().contains(mutation.id().offset())) + if (witnessedOffsets.get(localNodeId).contains(mutation.id().offset())) return false; // already witnessed; shouldn't get to this path often (duplicate mutation) unreconciledMutations.startWriting(mutation); @@ -143,7 +198,7 @@ public abstract class CoordinatorLog { int offset = mutation.id().offset(); // we've raced with another write, no need to do anything else - if (!getLocal().add(offset)) + if (!witnessedOffsets.get(localNodeId).add(offset)) return; unreconciledMutations.finishWriting(mutation); @@ -160,12 +215,12 @@ public abstract class CoordinatorLog } } - private boolean othersWitnessed(int offset, int exceptHostId) + private boolean othersWitnessed(int offset, int exceptNodeId) { for (int i = 0; i < participants.size(); ++i) { - int hostId = participants.get(i); - if (hostId != exceptHostId && !get(hostId).contains(offset)) + int nodeId = participants.get(i); + if (nodeId != exceptNodeId && !witnessedOffsets.get(nodeId).contains(offset)) return false; } return true; @@ -173,7 +228,7 @@ public abstract class CoordinatorLog protected boolean remoteReplicasWitnessed(int offset) { - return othersWitnessed(offset, localHostId); + return othersWitnessed(offset, localNodeId); } /** @@ -220,7 +275,7 @@ public abstract class CoordinatorLog lock.readLock().lock(); try { - into.add(Offsets.Immutable.difference(remoteOffsets, getLocal())); + into.add(Offsets.Immutable.difference(remoteOffsets, witnessedOffsets.get(localNodeId))); } finally { @@ -233,9 +288,11 @@ public abstract class CoordinatorLog lock.readLock().lock(); try { - remoteNodeIds.forEachInt(remoteNodeId -> { - Offsets missing = Offsets.Immutable.difference(get(remoteNodeId), localOffsets); - if (!missing.isEmpty()) into.add(remoteNodeId, missing); + remoteNodeIds.forEachInt(remoteNodeId -> + { + Offsets missing = Offsets.Immutable.difference(witnessedOffsets.get(remoteNodeId), localOffsets); + if (!missing.isEmpty()) + into.add(remoteNodeId, missing); }); } finally @@ -244,18 +301,12 @@ public abstract class CoordinatorLog } } - protected Offsets.Mutable get(int hostId) - { - return witnessedOffsets[participants.indexOf(hostId)]; - } - boolean isDurablyReconciled(CoordinatorLogOffsets logOffsets) { lock.readLock().lock(); try { - // TODO: reconciledOffsets not necessarily durable, update once durability is implemented - Offsets.RangeIterator durablyReconciled = reconciledOffsets.rangeIterator(); + Offsets.RangeIterator durablyReconciled = reconciledPersistedOffsets.rangeIterator(); Offsets.RangeIterator difference = Offsets.difference(logOffsets.offsets(logId.asLong()).rangeIterator(), durablyReconciled); return !difference.tryAdvance(); } @@ -265,17 +316,12 @@ public abstract class CoordinatorLog } } - protected Offsets.Mutable getLocal() - { - return witnessedOffsets[participants.indexOf(localHostId)]; - } - @Override public String toString() { return "CoordinatorLog{" + "logId=" + logId + - ", localHostId=" + localHostId + + ", localNodeId=" + localNodeId + ", participants=" + participants + '}'; } @@ -284,24 +330,31 @@ public abstract class CoordinatorLog { private final AtomicLong sequenceId = new AtomicLong(-1); - CoordinatorLogPrimary(int localHostId, CoordinatorLogId logId, Participants participants) + CoordinatorLogPrimary( + String keyspace, Range range, int localNodeId, CoordinatorLogId logId, Participants participants, + Node2OffsetsMap witnessedOffsets, Node2OffsetsMap persistedOffsets, UnreconciledMutations unreconciledMutations) { - super(localHostId, logId, participants); + super(keyspace, range, localNodeId, logId, participants, witnessedOffsets, persistedOffsets, unreconciledMutations); + } + + CoordinatorLogPrimary(String keyspace, Range range, int localNodeId, CoordinatorLogId logId, Participants participants) + { + super(keyspace, range, localNodeId, logId, participants); } @Override - void receivedWriteResponse(ShortMutationId mutationId, int fromHostId) + void receivedWriteResponse(ShortMutationId mutationId, int fromNodeId) { Preconditions.checkArgument(!mutationId.isNone()); - Preconditions.checkArgument(!Objects.equals(fromHostId, ClusterMetadata.current().myNodeId().id())); - logger.trace("witnessed remote mutation {} from {}", mutationId, fromHostId); + Preconditions.checkArgument(!Objects.equals(fromNodeId, ClusterMetadata.current().myNodeId().id())); + logger.trace("witnessed remote mutation {} from {}", mutationId, fromNodeId); lock.writeLock().lock(); try { - if (!get(fromHostId).add(mutationId.offset())) + if (!witnessedOffsets.get(fromNodeId).add(mutationId.offset())) return; // already witnessed; very uncommon but possible path - if (!getLocal().contains(mutationId.offset())) + if (!witnessedOffsets.get(localNodeId).contains(mutationId.offset())) return; // local host hasn't witnessed yet -> no cleanup needed if (remoteReplicasWitnessed(mutationId.offset())) @@ -318,9 +371,13 @@ public abstract class CoordinatorLog } } + @Nullable MutationId nextId() { - return new MutationId(logId.asLong(), nextSequenceId()); + long nextSequenceId = nextSequenceId(); + return nextSequenceId >= 0 + ? new MutationId(logId.asLong(), nextSequenceId) + : null; } private long nextSequenceId() @@ -331,6 +388,10 @@ public abstract class CoordinatorLog int prevOffset = MutationId.offset(prev); int prevTimestamp = MutationId.timestamp(prev); + // int overflow + if (prevOffset == MAX_OFFSET) + return -1; + int nextOffset = prevOffset + 1; int nextTimestamp = Math.max(prevTimestamp + 1, (int) (currentTimeMillis() / 1000L)); long next = MutationId.sequenceId(nextOffset, nextTimestamp); @@ -343,16 +404,131 @@ public abstract class CoordinatorLog static class CoordinatorLogReplica extends CoordinatorLog { - CoordinatorLogReplica(int localHostId, CoordinatorLogId logId, Participants participants) + CoordinatorLogReplica( + String keyspace, Range range, int localNodeId, CoordinatorLogId logId, Participants participants, + Node2OffsetsMap witnessedOffsets, Node2OffsetsMap persistedOffsets, UnreconciledMutations unreconciledMutations) { - super(localHostId, logId, participants); + super(keyspace, range, localNodeId, logId, participants, witnessedOffsets, persistedOffsets, unreconciledMutations); + } + + CoordinatorLogReplica(String keyspace, Range range, int localNodeId, CoordinatorLogId logId, Participants participants) + { + super(keyspace, range, localNodeId, logId, participants); } @Override - void receivedWriteResponse(ShortMutationId mutationId, int fromHostId) + void receivedWriteResponse(ShortMutationId mutationId, int fromNodeId) { // no-op } } + /* + * Persist to / load from system table. + */ + + private static final String INSERT_QUERY = + format("INSERT INTO %s.%s (keyspace_name, range_start, range_end, host_id, host_log_id, participants, witnessed_offsets, persisted_offsets) " + + "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.COORDINATOR_LOGS); + + void persistToSystemTable() + { + Map> witnessed = new Int2ObjectHashMap<>(); + Map> persisted = new Int2ObjectHashMap<>(); + + lock.readLock().lock(); + try + { + witnessedOffsets.convertToPrimitiveMap(witnessed); + persistedOffsets.convertToPrimitiveMap(persisted); + } + finally + { + lock.readLock().unlock(); + } + executeInternal(INSERT_QUERY, keyspace, range.left.toString(), range.right.toString(), logId.hostId, + logId.hostLogId, participants.asSet(), witnessed, persisted); + } + + void updateLogsInSystemTable() + { + Offsets.Mutable localWitnessed; + Map> witnessed = new Int2ObjectHashMap<>(); + Map> persisted = new Int2ObjectHashMap<>(); + + lock.readLock().lock(); + try + { + localWitnessed = Offsets.Mutable.copy(witnessedOffsets.get(localNodeId)); + + witnessedOffsets.convertToPrimitiveMap(witnessed); + persistedOffsets.convertToPrimitiveMap(persisted); + + persisted.put(localNodeId, witnessed.get(localNodeId)); + } + finally + { + lock.readLock().unlock(); + } + + executeInternal(INSERT_QUERY, keyspace, range.left.toString(), range.right.toString(), logId.hostId, + logId.hostLogId, participants.asSet(), witnessed, persisted); + + lock.writeLock().lock(); + try + { + persistedOffsets.set(localNodeId, localWitnessed); + reconciledPersistedOffsets.addAll(persistedOffsets.intersection()); + } + finally + { + lock.writeLock().unlock(); + } + } + + private static final String SELECT_QUERY = + format("SELECT * FROM %s.%s WHERE keyspace_name = ? AND range_start = ? AND range_end = ?", + SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.COORDINATOR_LOGS); + + static List loadFromSystemTable(String keyspace, Range range, int localNodeId) + { + ArrayList logs = new ArrayList<>(); + for (UntypedResultSet.Row row : executeInternal(SELECT_QUERY, keyspace, range.left.toString(), range.right.toString())) + { + int nodeId = row.getInt("host_id"); + int hostLogId = row.getInt("host_log_id"); + CoordinatorLogId logId = new CoordinatorLogId(nodeId, hostLogId); + Set participants = row.getFrozenSet("participants", Int32Type.instance); + Map> witnessedOffsets = + row.getMap("witnessed_offsets", Int32Type.instance, ListType.getInstance(Int32Type.instance, false)); + Map> persistedOffsets = + row.getMap("persisted_offsets", Int32Type.instance, ListType.getInstance(Int32Type.instance, false)); + Node2OffsetsMap witnessed = fromPrimitiveMap(logId, witnessedOffsets); + Node2OffsetsMap persisted = fromPrimitiveMap(logId, persistedOffsets); + UnreconciledMutations unreconciled = UnreconciledMutations.loadFromJournal(witnessed, localNodeId); + CoordinatorLog log = + CoordinatorLog.recreate(keyspace, range, localNodeId, logId, new Participants(participants), witnessed, persisted, unreconciled); + logs.add(log); + } + return logs; + } + + private static final String DELETE_QUERY = + format("DELETE FROM %s.%s WHERE keyspace_name = ? AND range_start = ? AND range_end = ? AND host_id = ? AND host_log_id = ?", + SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.COORDINATOR_LOGS); + + void deleteFromSystemTable() + { + executeInternal(DELETE_QUERY, keyspace, range.left.toString(), range.right.toString(), logId.hostId, logId.hostLogId); + } + + @VisibleForTesting + static void overrideMaxOffsetForTesting(int nexMaxOffset) + { + MAX_OFFSET = nexMaxOffset; + } + // don't make volatile unless it genuinely is an issue for some test, + // otherwise it should be *fine* as is, and slight overkill to make volatile + private static int MAX_OFFSET = Integer.MAX_VALUE; } diff --git a/src/java/org/apache/cassandra/replication/MutationTrackingService.java b/src/java/org/apache/cassandra/replication/MutationTrackingService.java index 7ec6758707..6dc1bcc963 100644 --- a/src/java/org/apache/cassandra/replication/MutationTrackingService.java +++ b/src/java/org/apache/cassandra/replication/MutationTrackingService.java @@ -17,6 +17,7 @@ */ package org.apache.cassandra.replication; +import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.Iterator; @@ -26,10 +27,9 @@ import java.util.Optional; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiConsumer; import java.util.function.Consumer; -import java.util.function.IntSupplier; +import java.util.function.LongSupplier; import javax.annotation.Nonnull; import javax.annotation.Nullable; @@ -40,10 +40,11 @@ import com.google.common.base.Preconditions; import org.agrona.collections.IntArrayList; import org.agrona.collections.IntHashSet; import org.apache.cassandra.concurrent.ScheduledExecutorPlus; -import org.apache.cassandra.concurrent.Shutdownable; +import org.apache.cassandra.cql3.UntypedResultSet; import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.Mutation; import org.apache.cassandra.db.PartitionPosition; +import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.dht.AbstractBounds; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Splitter; @@ -55,10 +56,13 @@ import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.net.Verb; import org.apache.cassandra.schema.KeyspaceMetadata; +import org.apache.cassandra.schema.ReplicationParams; import org.apache.cassandra.schema.Schema; +import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.service.reads.tracked.TrackedLocalReads; import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.membership.NodeId; import org.apache.cassandra.tcm.ownership.ReplicaGroups; import org.apache.cassandra.tcm.ownership.VersionedEndpoints; import org.apache.cassandra.utils.FBUtilities; @@ -66,14 +70,17 @@ import org.apache.cassandra.utils.FBUtilities; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import static java.util.concurrent.TimeUnit.MINUTES; +import static java.lang.String.format; import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory; import static org.apache.cassandra.concurrent.ExecutorFactory.SimulatorSemantics.NORMAL; +import static org.apache.cassandra.cql3.QueryProcessor.executeInternal; // TODO (expected): persistence (handle restarts) // TODO (expected): handle topology changes public class MutationTrackingService { + private static final ScheduledExecutorPlus executor = executorFactory().scheduled("Mutation-Tracking-Service", NORMAL); + /** * Split ranges into this many shards. * @@ -89,6 +96,7 @@ public class MutationTrackingService private final ConcurrentHashMap log2ShardMap = new ConcurrentHashMap<>(); private final ReplicatedOffsetsBroadcaster offsetsBroadcaster = new ReplicatedOffsetsBroadcaster(); + private final LogStatePersister offsetsPersister = new LogStatePersister(); private final ActiveLogReconciler activeReconciler = new ActiveLogReconciler(); private final IncomingMutations incomingMutations = new IncomingMutations(); @@ -104,13 +112,16 @@ public class MutationTrackingService if (started) return; - logger.info("Starting replication tracking service"); + prevHostLogId = loadHostLogIdFromSystemTable(); - for (KeyspaceMetadata keyspace : metadata.schema.getKeyspaces()) - if (keyspace.useMutationTracking()) - keyspaceShards.put(keyspace.name, KeyspaceShards.make(keyspace, metadata, this::nextHostLogId, this::onNewLog)); + logger.info("Starting mutation tracking service. Previous host log id: {}", prevHostLogId); + + if (metadata.myNodeId() != null) + for (KeyspaceShards ks : KeyspaceShards.loadFromSystemTables(metadata, this::nextLogId, this::onNewLog)) + keyspaceShards.put(ks.keyspace, ks); offsetsBroadcaster.start(); + offsetsPersister.start(); ExpiredStatePurger.instance.register(incomingMutations); @@ -124,9 +135,10 @@ public class MutationTrackingService public void shutdownBlocking() throws InterruptedException { - offsetsBroadcaster.shutdown(); - offsetsBroadcaster.awaitTermination(1, TimeUnit.MINUTES); + // TODO: FIXME activeReconciler.shutdownBlocking(); + executor.shutdown(); + executor.awaitTermination(1, TimeUnit.MINUTES); ExpiredStatePurger.instance.shutdownBlocking(); } @@ -165,9 +177,9 @@ public class MutationTrackingService activeReconciler.schedule(mutationId, onHost, ActiveLogReconciler.Priority.REGULAR); } - public void updateReplicatedOffsets(String keyspace, Range range, List offsets, InetAddressAndPort onHost) + public void updateReplicatedOffsets(String keyspace, Range range, List offsets, boolean durable, InetAddressAndPort onHost) { - getOrCreateShards(keyspace).updateReplicatedOffsets(range, offsets, onHost); + getOrCreateShards(keyspace).updateReplicatedOffsets(range, offsets, durable, onHost); } public boolean startWriting(Mutation mutation) @@ -264,41 +276,41 @@ public class MutationTrackingService ClusterMetadata csm = ClusterMetadata.current(); KeyspaceMetadata ksm = csm.schema.getKeyspaceMetadata(keyspace); - return keyspaceShards.computeIfAbsent(keyspace, ignore -> KeyspaceShards.make(ksm, csm, this::nextHostLogId, this::onNewLog)); + return keyspaceShards.computeIfAbsent(keyspace, ignore -> KeyspaceShards.make(ksm, csm, this::nextLogId, this::onNewLog)); } - // TODO (expected): durability - int nextHostLogId() + private long nextLogId() { - return nextHostLogId.incrementAndGet(); + NodeId nodeId = ClusterMetadata.current().myNodeId(); + Preconditions.checkNotNull(nodeId); + return CoordinatorLogId.asLong(nodeId.id(), nextHostLogId()); } - private final AtomicInteger nextHostLogId = new AtomicInteger(); - public boolean isDurablyReconciled(String keyspace, ImmutableCoordinatorLogOffsets logOffsets) + /* + * Allocate and persist the next host log id. + * We only do this on startup and when rotating logs. + */ + private int nextHostLogId() + { + int nextHostLogId = ++prevHostLogId; + persistHostLogIdToSystemTable(nextHostLogId); + return nextHostLogId; + } + private int prevHostLogId; + + public boolean isDurablyReconciled(ImmutableCoordinatorLogOffsets logOffsets) { // Could pass through SSTable bounds to exclude shards for non-overlapping ranges, but this will mostly be // called on flush for L0 SSTables with wide bounds. - - KeyspaceShards shards = keyspaceShards.get(keyspace); - if (shards == null) - { - logger.debug("Could not find shards for keyspace {}", keyspace); - return false; - } - for (Long logId : logOffsets) { - CoordinatorLogId coordinatorLogId = new CoordinatorLogId(logId); - CoordinatorLog log = shards.logs.get(coordinatorLogId); - if (log == null) - { - logger.warn("Could not determine lifecycle for unknown logId {}, not marking as durably reconciled", coordinatorLogId); - return false; - } - if (!log.isDurablyReconciled(logOffsets)) + Shard shard = getShardNullable(new CoordinatorLogId(logId)); + if (shard == null) + throw new IllegalStateException("Could not find shard for logId " + logId); + + if (!shard.isDurablyReconciled(logId, logOffsets)) return false; } - return true; } @@ -308,17 +320,15 @@ public class MutationTrackingService log2ShardMap.put(log.logId, shard); } - private static class KeyspaceShards implements Shard.Subscriber + private static class KeyspaceShards { private final String keyspace; private final Map, Shard> shards; private final ReplicaGroups groups; - private final BiConsumer onNewLog; private transient final Map, Shard> ppShards; - private transient final Map logs; - static KeyspaceShards make(KeyspaceMetadata keyspace, ClusterMetadata cluster, IntSupplier logIdProvider, BiConsumer onNewLog) + static KeyspaceShards make(KeyspaceMetadata keyspace, ClusterMetadata cluster, LongSupplier logIdProvider, BiConsumer onNewLog) { Preconditions.checkArgument(keyspace.params.replicationType.isTracked()); @@ -341,26 +351,23 @@ public class MutationTrackingService for (Range tokenRange : ranges) { - shards.put(tokenRange, new Shard(keyspace.name, tokenRange, cluster.myNodeId().id(), participants, forRange.lastModified(), logIdProvider, onNewLog)); + shards.put(tokenRange, new Shard(cluster.myNodeId().id(), keyspace.name, tokenRange, participants, logIdProvider, onNewLog)); groups.put(tokenRange, forRange.map(original -> original.withRange(tokenRange))); } }); - return new KeyspaceShards(keyspace.name, shards, new ReplicaGroups(groups), onNewLog); + KeyspaceShards keyspaceShards = new KeyspaceShards(keyspace.name, shards, new ReplicaGroups(groups)); + keyspaceShards.persistToSystemTables(); + return keyspaceShards; } - KeyspaceShards(String keyspace, Map, Shard> shards, ReplicaGroups groups, BiConsumer onNewLog) + KeyspaceShards(String keyspace, Map, Shard> shards, ReplicaGroups groups) { this.keyspace = keyspace; this.shards = shards; this.groups = groups; - this.onNewLog = onNewLog; - this.logs = new HashMap<>(); HashMap, Shard> ppShards = new HashMap<>(); - shards.forEach((range, shard) -> { - ppShards.put(Range.makeRowRange(range), shard); - shard.addSubscriber(this); - }); + shards.forEach((range, shard) -> ppShards.put(Range.makeRowRange(range), shard)); this.ppShards = ppShards; } @@ -369,9 +376,9 @@ public class MutationTrackingService return lookUp(token).nextId(); } - void updateReplicatedOffsets(Range range, List offsets, InetAddressAndPort onHost) + void updateReplicatedOffsets(Range range, List offsets, boolean durable, InetAddressAndPort onHost) { - shards.get(range).updateReplicatedOffsets(offsets, onHost); + shards.get(range).updateReplicatedOffsets(offsets, durable, onHost); } boolean startWriting(Mutation mutation) @@ -430,66 +437,105 @@ public class MutationTrackingService return shards.get(groups.forRange(token).range()); } - @Override - public void onLogCreation(CoordinatorLog log) + void persistToSystemTables() { - logger.debug("Indexing created log {}", log); - logs.put(log.logId, log); + for (Shard shard : shards.values()) shard.persistToSystemTables(); } - @Override - public void onSubscribe(CoordinatorLog currentLog) + static List loadFromSystemTables(ClusterMetadata cluster, LongSupplier logIdProvider, BiConsumer onNewLog) { - logger.debug("Indexing current log {}", currentLog); - logs.put(currentLog.logId, currentLog); + Map, Shard>> groupedShards = new HashMap<>(); + for (Shard shard : Shard.loadFromSystemTables(cluster.myNodeId().id(), logIdProvider, onNewLog)) + groupedShards.computeIfAbsent(shard.keyspace, k -> new HashMap<>()).put(shard.range, shard); + List keyspaceShards = new ArrayList<>(); + for (Map.Entry, Shard>> entry : groupedShards.entrySet()) + { + ReplicationParams params = cluster.schema.getKeyspaceMetadata(entry.getKey()).params.replication; + ReplicaGroups originalGroups = cluster.placements.get(params).writes; // prior to splitting + + Map, VersionedEndpoints.ForRange> splitGroups = new HashMap<>(); + for (Range splitRange : entry.getValue().keySet()) + splitGroups.put(splitRange, originalGroups.matchRange(splitRange)); + + keyspaceShards.add(new KeyspaceShards(entry.getKey(), entry.getValue(), new ReplicaGroups(splitGroups))); + } + return keyspaceShards; } } - // TODO (later): a more intelligent heuristic for offsets included in broadcasts - private static class ReplicatedOffsetsBroadcaster implements Runnable, Shutdownable - { - private static final ScheduledExecutorPlus executor = - executorFactory().scheduled("Replicated-Offsets-Broadcaster", NORMAL); + private static final String HOST_LOG_ID_KEY = "local"; + private static final String INSERT_QUERY = + format("INSERT INTO %s.%s (key, host_log_id) VALUES (?, ?)", + SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.HOST_LOG_ID); + + static void persistHostLogIdToSystemTable(int hostLogId) + { + executeInternal(INSERT_QUERY, HOST_LOG_ID_KEY, hostLogId); + } + + private static final String SELECT_QUERY = + format("SELECT * FROM %s.%s WHERE key = ?", SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.HOST_LOG_ID); + + static int loadHostLogIdFromSystemTable() + { + UntypedResultSet rows = executeInternal(SELECT_QUERY, HOST_LOG_ID_KEY); + if (rows.isEmpty()) + return 0; + return rows.one().getInt("host_log_id"); + } + + // TODO (later): a more intelligent heuristic for offsets included in broadcasts + private static class ReplicatedOffsetsBroadcaster + { // TODO (later): a more intelligent heuristic for scheduling broadcasts - private static final long BROADCAST_INTERVAL_MILLIS = 200; + private static final long TRANSIENT_BROADCAST_INTERVAL_MILLIS = 200; + private static final long DURABLE_BROADCAST_INTERVAL_MILLIS = 60_000; void start() { - executor.scheduleWithFixedDelay(this, BROADCAST_INTERVAL_MILLIS, BROADCAST_INTERVAL_MILLIS, TimeUnit.MILLISECONDS); + executor.scheduleWithFixedDelay(() -> run(false), + TRANSIENT_BROADCAST_INTERVAL_MILLIS, + TRANSIENT_BROADCAST_INTERVAL_MILLIS, + TimeUnit.MILLISECONDS); + executor.scheduleWithFixedDelay(() -> run(true), + DURABLE_BROADCAST_INTERVAL_MILLIS, + DURABLE_BROADCAST_INTERVAL_MILLIS, + TimeUnit.MILLISECONDS); } - @Override - public boolean isTerminated() + public void run(boolean durable) { - return executor.isTerminated(); + MutationTrackingService.instance.forEachKeyspace(ks -> run(ks, durable)); } - @Override - public void shutdown() + private void run(KeyspaceShards shards, boolean durable) { - executor.shutdown(); + shards.forEachShard(sh -> run(sh, durable)); } - @Override - public Object shutdownNow() + private void run(Shard shard, boolean durable) { - return executor.shutdownNow(); - } + BroadcastLogOffsets replicatedOffsets = shard.collectReplicatedOffsets(durable); + if (replicatedOffsets.isEmpty()) + return; - @Override - public boolean awaitTermination(long timeout, TimeUnit units) throws InterruptedException - { - return executor.awaitTermination(timeout, units); - } + Message message = Message.out(Verb.BROADCAST_LOG_OFFSETS, replicatedOffsets); - public void shutdownBlocking() throws InterruptedException + for (InetAddressAndPort target : shard.remoteReplicas()) + if (FailureDetector.instance.isAlive(target)) + MessagingService.instance().send(message, target); + } + } + + private static class LogStatePersister implements Runnable + { + // TODO (expected): consider a different interval + private static final long PERSIST_INTERVAL_MINUTES = 1; + + void start() { - if (!executor.isTerminated()) - { - executor.shutdown(); - executor.awaitTermination(1, MINUTES); - } + executor.scheduleWithFixedDelay(this, PERSIST_INTERVAL_MINUTES, PERSIST_INTERVAL_MINUTES, TimeUnit.MINUTES); } @Override @@ -505,22 +551,21 @@ public class MutationTrackingService private void run(Shard shard) { - BroadcastLogOffsets replicatedOffsets = shard.collectReplicatedOffsets(); - if (replicatedOffsets.isEmpty()) - return; - - Message message = Message.out(Verb.BROADCAST_LOG_OFFSETS, replicatedOffsets); - - for (InetAddressAndPort target : shard.remoteReplicas()) - if (FailureDetector.instance.isAlive(target)) - MessagingService.instance().send(message, target); + shard.updateLogsInSystemTable(); } } + @VisibleForTesting + public void persistLogStateForTesting() + { + offsetsPersister.run(); + } + @VisibleForTesting public void broadcastOffsetsForTesting() { - offsetsBroadcaster.run(); + offsetsBroadcaster.run(false); + offsetsBroadcaster.run(true); } @VisibleForTesting diff --git a/src/java/org/apache/cassandra/replication/Node2OffsetsMap.java b/src/java/org/apache/cassandra/replication/Node2OffsetsMap.java index 12fa18c98a..1001d855cd 100644 --- a/src/java/org/apache/cassandra/replication/Node2OffsetsMap.java +++ b/src/java/org/apache/cassandra/replication/Node2OffsetsMap.java @@ -17,6 +17,14 @@ */ package org.apache.cassandra.replication; +import java.util.Iterator; +import java.util.List; +import java.util.Map; + +import javax.annotation.Nonnull; + +import com.google.common.base.Preconditions; + import org.agrona.collections.Int2ObjectHashMap; import org.agrona.collections.IntObjConsumer; @@ -25,24 +33,93 @@ import org.agrona.collections.IntObjConsumer; */ public class Node2OffsetsMap { - private final Int2ObjectHashMap map = new Int2ObjectHashMap<>(); + private final Int2ObjectHashMap offsetsMap; + + public Node2OffsetsMap() + { + offsetsMap = new Int2ObjectHashMap<>(8, 0.65f, false); + } + + public static Node2OffsetsMap forParticipants(CoordinatorLogId logId, Participants participants) + { + Node2OffsetsMap map = new Node2OffsetsMap(); + for (int i = 0; i < participants.size(); i++) + map.set(participants.get(i), new Offsets.Mutable(logId)); + return map; + } + + void set(int node, Offsets.Mutable offsets) + { + offsetsMap.put(node, offsets); + } + + @Nonnull + Offsets.Mutable get(int node) + { + return Preconditions.checkNotNull(offsetsMap.get(node)); + } + + Offsets.Mutable intersection() + { + Iterator iter = offsetsMap.values().iterator(); + + Preconditions.checkArgument(iter.hasNext()); + if (offsetsMap.size() == 1) + return Offsets.Mutable.copy(iter.next()); + + Offsets.Mutable intersection = iter.next(); + while (iter.hasNext()) + intersection = Offsets.Mutable.intersection(intersection, iter.next()); + return intersection; + } public void add(int node, Offsets offsets) { - Offsets.Mutable current = (Offsets.Mutable) map.get(node); + Offsets.Mutable current = offsetsMap.get(node); if (current != null) current.addAll(offsets); else - map.put(node, Offsets.Mutable.copy(offsets)); + offsetsMap.put(node, Offsets.Mutable.copy(offsets)); } - public void forEach(IntObjConsumer consumer) + public void forEach(IntObjConsumer consumer) { - map.forEachInt(consumer); + offsetsMap.forEachInt(consumer); } public void clear() { - map.clear(); + offsetsMap.clear(); + } + + public int size() + { + return offsetsMap.size(); + } + + void convertToPrimitiveMap(Map> into) + { + for (Int2ObjectHashMap.EntryIterator iter = offsetsMap.entrySet().iterator(); iter.hasNext();) + { + iter.next(); + into.put(iter.getIntKey(), iter.getValue().asList()); + } + } + + static Node2OffsetsMap fromPrimitiveMap(CoordinatorLogId logId, Map> from) + { + Node2OffsetsMap map = new Node2OffsetsMap(); + for (Map.Entry> entry : from.entrySet()) + map.set(entry.getKey(), Offsets.fromList(logId, entry.getValue())); + return map; + } + + @Override + public boolean equals(Object o) + { + if (!(o instanceof Node2OffsetsMap)) + return false; + Node2OffsetsMap that = (Node2OffsetsMap) o; + return this.offsetsMap.equals(that.offsetsMap); } } diff --git a/src/java/org/apache/cassandra/replication/Offsets.java b/src/java/org/apache/cassandra/replication/Offsets.java index fc080452a7..dfb4f3f404 100644 --- a/src/java/org/apache/cassandra/replication/Offsets.java +++ b/src/java/org/apache/cassandra/replication/Offsets.java @@ -21,6 +21,7 @@ import com.google.common.base.Function; import com.google.common.base.Objects; import com.google.common.base.Preconditions; +import org.agrona.collections.IntArrayList; import org.apache.cassandra.db.Digest; import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.IVersionedSerializer; @@ -34,6 +35,7 @@ import java.io.IOException; import java.util.Arrays; import java.util.Collection; import java.util.Iterator; +import java.util.List; public abstract class Offsets implements Iterable { @@ -200,6 +202,19 @@ public abstract class Offsets implements Iterable digest.updateWithInt(bounds[i]); } + List asList() + { + return new IntArrayList(Arrays.copyOf(bounds, size), size, IntArrayList.DEFAULT_NULL_VALUE); + } + + static Mutable fromList(CoordinatorLogId logId, List list) + { + int[] bounds = new int[list.size()]; + for (int i = 0; i < list.size(); i++) + bounds[i] = list.get(i); + return new Mutable(logId, bounds, bounds.length); + } + static abstract class AbstractMutable> extends Offsets implements OffsetReciever { AbstractMutable(RangeIterator rangeIterator) @@ -380,6 +395,11 @@ public abstract class Offsets implements Iterable // the offset is before all existing ranges, in-between two, or after all exsiting ranges } + public void add(int... offsets) + { + for (int offset : offsets) add(offset); + } + private enum AddAction { INSERT, MOVE, INCLUDE; @@ -673,6 +693,19 @@ public abstract class Offsets implements Iterable { return Offsets.intersection(a, b, adaptor); } + + public static Mutable intersection(Offsets.Mutable... offsets) + { + Preconditions.checkArgument(offsets.length > 0); + + if (offsets.length == 1) + return copy(offsets[0]); + + Mutable intersection = offsets[0]; + for (int i = 1; i < offsets.length; i++) + intersection = intersection(intersection, offsets[i]); + return intersection; + } } public static class Immutable extends Offsets diff --git a/src/java/org/apache/cassandra/replication/Participants.java b/src/java/org/apache/cassandra/replication/Participants.java index f245797586..89494f240c 100644 --- a/src/java/org/apache/cassandra/replication/Participants.java +++ b/src/java/org/apache/cassandra/replication/Participants.java @@ -19,6 +19,9 @@ package org.apache.cassandra.replication; import java.util.Arrays; import java.util.Collection; +import java.util.Set; + +import org.agrona.collections.IntHashSet; public class Participants { @@ -63,4 +66,21 @@ public class Participants { return Arrays.toString(hosts); } + + @Override + public boolean equals(Object o) + { + if (!(o instanceof Participants)) + return false; + Participants that = (Participants) o; + return Arrays.equals(this.hosts, that.hosts); + } + + Set asSet() + { + IntHashSet set = new IntHashSet(hosts.length); + for (int host : hosts) + set.add(host); + return set; + } } diff --git a/src/java/org/apache/cassandra/replication/Shard.java b/src/java/org/apache/cassandra/replication/Shard.java index 6588bede09..6f198e686d 100644 --- a/src/java/org/apache/cassandra/replication/Shard.java +++ b/src/java/org/apache/cassandra/replication/Shard.java @@ -18,72 +18,100 @@ package org.apache.cassandra.replication; import java.util.ArrayList; +import java.util.Collections; import java.util.List; +import java.util.Set; import java.util.function.BiConsumer; -import java.util.function.IntSupplier; +import java.util.function.LongSupplier; import javax.annotation.Nonnull; import com.google.common.base.Preconditions; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import org.agrona.collections.IntArrayList; +import org.apache.cassandra.cql3.UntypedResultSet; import org.apache.cassandra.db.Mutation; import org.apache.cassandra.db.PartitionPosition; +import org.apache.cassandra.db.SystemKeyspace; +import org.apache.cassandra.db.marshal.Int32Type; import org.apache.cassandra.dht.AbstractBounds; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.replication.CoordinatorLog.CoordinatorLogPrimary; +import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.tcm.ClusterMetadata; -import org.apache.cassandra.tcm.Epoch; import org.apache.cassandra.tcm.membership.NodeId; import org.jctools.maps.NonBlockingHashMapLong; +import static java.lang.String.format; +import static org.apache.cassandra.cql3.QueryProcessor.executeInternal; + public class Shard { - private final String keyspace; - private final Range tokenRange; - private final int localHostId; - private final Participants participants; - private final Epoch sinceEpoch; + private static final Logger logger = LoggerFactory.getLogger(Shard.class); + + final int localNodeId; + final String keyspace; + final Range range; + final Participants participants; + + private final LongSupplier logIdProvider; private final BiConsumer onNewLog; private final NonBlockingHashMapLong logs; - // TODO (expected): add support for log rotation - private final CoordinatorLog.CoordinatorLogPrimary currentLocalLog; + private volatile CoordinatorLogPrimary currentLocalLog; - private final List subscribers = new ArrayList<>(); - - public interface Subscriber - { - default void onLogCreation(CoordinatorLog log) {} - default void onSubscribe(CoordinatorLog currentLog) {} - } - - Shard(String keyspace, - Range tokenRange, - int localHostId, + Shard(int localNodeId, + String keyspace, + Range range, Participants participants, - Epoch sinceEpoch, - IntSupplier logIdProvider, + List logs, + LongSupplier logIdProvider, BiConsumer onNewLog) { - Preconditions.checkArgument(participants.contains(localHostId)); + Preconditions.checkArgument(participants.contains(localNodeId)); + this.localNodeId = localNodeId; this.keyspace = keyspace; - this.tokenRange = tokenRange; - this.localHostId = localHostId; + this.range = range; this.participants = participants; - this.sinceEpoch = sinceEpoch; + this.logIdProvider = logIdProvider; this.logs = new NonBlockingHashMapLong<>(); this.onNewLog = onNewLog; - this.currentLocalLog = startNewLog(localHostId, logIdProvider.getAsInt(), participants); - CoordinatorLogId logId = currentLocalLog.logId; - Preconditions.checkArgument(!logId.isNone()); - logs.put(logId.asLong(), currentLocalLog); + for (CoordinatorLog log : logs) + { + this.logs.put(log.logId.asLong(), log); + onNewLog.accept(Shard.this, log); + } + this.currentLocalLog = createNewPrimaryLog(); + } + + Shard(int localNodeId, String keyspace, Range range, Participants participants, LongSupplier logIdProvider, BiConsumer onNewLog) + { + this(localNodeId, keyspace, range, participants, Collections.emptyList(), logIdProvider, onNewLog); } MutationId nextId() { - return currentLocalLog.nextId(); + MutationId nextId = currentLocalLog.nextId(); + if (nextId != null) + return nextId; + return maybeRotateLocalLogAndGetNextId(); + } + + // if ids overflow, we need to rotate the local log + synchronized private MutationId maybeRotateLocalLogAndGetNextId() + { + MutationId nextId = currentLocalLog.nextId(); + if (nextId != null) // another thread got to rotate before us + return nextId; + CoordinatorLogId oldLogId = currentLocalLog.logId; + currentLocalLog = createNewPrimaryLog(); + logger.info("Rotated primary log for {}/{} from {} to {}", keyspace, range, oldLogId, currentLocalLog.logId); + return nextId(); } void receivedWriteResponse(ShortMutationId mutationId, InetAddressAndPort fromHost) @@ -92,11 +120,11 @@ public class Shard getOrCreate(mutationId).receivedWriteResponse(mutationId, fromHostId); } - void updateReplicatedOffsets(List offsets, InetAddressAndPort onHost) + void updateReplicatedOffsets(List offsets, boolean durable, InetAddressAndPort onHost) { int onHostId = ClusterMetadata.current().directory.peerId(onHost).id(); for (Offsets logOffsets : offsets) - getOrCreate(logOffsets.logId()).updateReplicatedOffsets(logOffsets, onHostId); + getOrCreate(logOffsets.logId()).updateReplicatedOffsets(logOffsets, durable, onHostId); } boolean startWriting(Mutation mutation) @@ -143,39 +171,31 @@ public class Shard for (int i = 0, size = participants.size(); i < size; ++i) { int hostId = participants.get(i); - if (hostId != localHostId) + if (hostId != localNodeId) replicas.add(ClusterMetadata.current().directory.endpoint(new NodeId(hostId))); } return replicas; } + boolean isDurablyReconciled(long logId, CoordinatorLogOffsets logOffsets) + { + return logs.get(logId).isDurablyReconciled(logOffsets); + } + /** * Collects replicated offsets for the logs owned by this coordinator on this shard. */ - BroadcastLogOffsets collectReplicatedOffsets() + BroadcastLogOffsets collectReplicatedOffsets(boolean durable) { List offsets = new ArrayList<>(); for (CoordinatorLog log : logs.values()) { - Offsets.Immutable logOffsets = log.collectReplicatedOffsets(); + Offsets.Immutable logOffsets = log.collectReplicatedOffsets(durable); if (logOffsets != null) offsets.add(logOffsets); } - return new BroadcastLogOffsets(keyspace, tokenRange, offsets); - } - - /** - * Creates a new coordinator log for this host. Primarily on Shard init (node startup or topology change). - * Also on keyspace creation. - */ - private CoordinatorLog.CoordinatorLogPrimary startNewLog(int localHostId, int hostLogId, Participants participants) - { - CoordinatorLogId logId = new CoordinatorLogId(localHostId, hostLogId); - CoordinatorLog.CoordinatorLogPrimary log = - new CoordinatorLog.CoordinatorLogPrimary(localHostId, logId, participants); - onNewLog.accept(this, log); - return log; + return new BroadcastLogOffsets(keyspace, range, offsets, durable); } private CoordinatorLog getOrCreate(Mutation mutation) @@ -203,27 +223,63 @@ public class Shard private CoordinatorLog getOrCreate(long logId) { CoordinatorLog log = logs.get(logId); - if (log != null) - return log; - CoordinatorLog newLog = logs.computeIfAbsent(logId, ignore -> CoordinatorLog.create(localHostId, new CoordinatorLogId(logId), participants)); - onNewLog.accept(this, newLog); - for (Subscriber subscriber : subscribers) - subscriber.onLogCreation(newLog); - return newLog; + return log != null ? log : createNewLog(logId); } - public void addSubscriber(Subscriber subscriber) + /** + * Creates a new coordinator log for this host. Primarily on Shard init (node startup or topology change) and on keyspace creation. + */ + private CoordinatorLog createNewLog(long logId) { - subscriber.onSubscribe(currentLocalLog); - subscribers.add(subscriber); + CoordinatorLog next = CoordinatorLog.create(keyspace, range, localNodeId, new CoordinatorLogId(logId), participants); + CoordinatorLog prev = logs.putIfAbsent(logId, next); + if (null == prev) onNewLog.accept(this, next); + return null != prev ? prev : next; } - private CoordinatorLog create(long logId) + private CoordinatorLogPrimary createNewPrimaryLog() { - CoordinatorLog log = CoordinatorLog.create(localHostId, new CoordinatorLogId(logId), participants); - onNewLog.accept(this, log); - for (Subscriber subscriber : subscribers) - subscriber.onLogCreation(log); - return log; + return (CoordinatorLogPrimary) createNewLog(logIdProvider.getAsLong()); + } + + /* + * Persist to / load from system table. + */ + + private static final String INSERT_QUERY = + format("INSERT INTO %s.%s (keyspace_name, range_start, range_end, participants) VALUES (?, ?, ?, ?)", + SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.SHARDS); + + void persistToSystemTables() + { + executeInternal(INSERT_QUERY, keyspace, range.left.toString(), range.right.toString(), participants.asSet()); + for (CoordinatorLog log : logs.values()) + log.persistToSystemTable(); + } + + void updateLogsInSystemTable() + { + for (CoordinatorLog log : logs.values()) + log.updateLogsInSystemTable(); + } + + private static final String SELECT_QUERY = + format("SELECT * FROM %s.%s", SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.SHARDS); + + static ArrayList loadFromSystemTables(int localNodeId, LongSupplier logIdProvider, BiConsumer onNewLog) + { + Token.TokenFactory factory = ClusterMetadata.current().partitioner.getTokenFactory(); + ArrayList shards = new ArrayList<>(); + for (UntypedResultSet.Row row : executeInternal(SELECT_QUERY)) + { + String keyspace = row.getString("keyspace_name"); + String rangeStart = row.getString("range_start"); + String rangeEnd = row.getString("range_end"); + Range range = new Range<>(factory.fromString(rangeStart), factory.fromString(rangeEnd)); + Set participants = row.getFrozenSet("participants", Int32Type.instance); + List logs = CoordinatorLog.loadFromSystemTable(keyspace, range, localNodeId); + shards.add(new Shard(localNodeId, keyspace, range, new Participants(participants), logs, logIdProvider, onNewLog)); + } + return shards; } } diff --git a/src/java/org/apache/cassandra/replication/UnreconciledMutations.java b/src/java/org/apache/cassandra/replication/UnreconciledMutations.java index 5a6e9d1f55..63dff77349 100644 --- a/src/java/org/apache/cassandra/replication/UnreconciledMutations.java +++ b/src/java/org/apache/cassandra/replication/UnreconciledMutations.java @@ -23,6 +23,7 @@ import java.util.Set; import java.util.SortedSet; import java.util.TreeSet; +import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import com.google.common.collect.Sets; @@ -197,4 +198,41 @@ public class UnreconciledMutations return collect(statesSet.subSet(start, end), tableId, includePending, into); } } + + @VisibleForTesting + boolean equalsForTesting(UnreconciledMutations other) + { + return this.statesMap.equals(other.statesMap) && this.statesSet.equals(other.statesSet); + } + + @VisibleForTesting + void addDirectly(Mutation mutation) + { + Entry entry = Entry.create(mutation); + entry.visibility = Visibility.VISIBLE; + statesMap.put(entry.offset, entry); + statesSet.add(entry); + } + + static UnreconciledMutations loadFromJournal(Node2OffsetsMap witnessedOffsets, int localNodeId) + { + UnreconciledMutations result = new UnreconciledMutations(); + + Offsets.Mutable witnessed = witnessedOffsets.get(localNodeId); + Offsets.Mutable reconciled = witnessedOffsets.intersection(); + + // difference between locally witnessed offsets and fully reconciled ones is all the ids + // that need to be loaded into UnreconciledMutations index + Offsets.RangeIterator iter = Offsets.difference(witnessed.rangeIterator(), reconciled.rangeIterator()); + while (iter.tryAdvance()) + { + for (int offset = iter.start(), end = iter.end(); offset <= end; offset++) + { + ShortMutationId id = new ShortMutationId(witnessed.logId, offset); + result.addDirectly(MutationJournal.instance.read(id)); + } + } + + return result; + } } diff --git a/src/java/org/apache/cassandra/tools/OfflineClusterMetadataDump.java b/src/java/org/apache/cassandra/tools/OfflineClusterMetadataDump.java index d0b9b54fea..8304ae9d14 100644 --- a/src/java/org/apache/cassandra/tools/OfflineClusterMetadataDump.java +++ b/src/java/org/apache/cassandra/tools/OfflineClusterMetadataDump.java @@ -285,10 +285,10 @@ public class OfflineClusterMetadataDump implements Runnable } // Find and import SSTables for metadata_snapshots - Set snapshotTablePaths = findTablePaths(SystemKeyspace.SNAPSHOT_TABLE_NAME, SchemaConstants.SYSTEM_KEYSPACE_NAME); + Set snapshotTablePaths = findTablePaths(SystemKeyspace.METADATA_SNAPSHOTS, SchemaConstants.SYSTEM_KEYSPACE_NAME); if (!snapshotTablePaths.isEmpty()) { - ColumnFamilyStore snapshotCfs = ks.getColumnFamilyStore(SystemKeyspace.SNAPSHOT_TABLE_NAME); + ColumnFamilyStore snapshotCfs = ks.getColumnFamilyStore(SystemKeyspace.METADATA_SNAPSHOTS); snapshotCfs.importNewSSTables(snapshotTablePaths, false, false, false, false, false, false, true); if (verbose) { diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/SystemKeyspaceStorageTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/SystemKeyspaceStorageTest.java index 3e158ac41c..e3958eefd3 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/log/SystemKeyspaceStorageTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/log/SystemKeyspaceStorageTest.java @@ -39,7 +39,7 @@ import org.apache.cassandra.tcm.log.LogState; import org.apache.cassandra.tcm.transformations.CustomTransformation; import org.apache.cassandra.utils.FBUtilities; -import static org.apache.cassandra.db.SystemKeyspace.SNAPSHOT_TABLE_NAME; +import static org.apache.cassandra.db.SystemKeyspace.METADATA_SNAPSHOTS; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -184,7 +184,7 @@ public class SystemKeyspaceStorageTest extends CoordinatorPathTestBase public static void deleteSnapshot(long epoch) { - String query = String.format("DELETE FROM %s.%s WHERE epoch = ?", SchemaConstants.SYSTEM_KEYSPACE_NAME, SNAPSHOT_TABLE_NAME); + String query = String.format("DELETE FROM %s.%s WHERE epoch = ?", SchemaConstants.SYSTEM_KEYSPACE_NAME, METADATA_SNAPSHOTS); QueryProcessor.executeInternal(query, epoch); } } diff --git a/test/unit/org/apache/cassandra/cql3/SystemKeyspaceQueryTest.java b/test/unit/org/apache/cassandra/cql3/SystemKeyspaceQueryTest.java index 8da29df678..f5789d01f5 100644 --- a/test/unit/org/apache/cassandra/cql3/SystemKeyspaceQueryTest.java +++ b/test/unit/org/apache/cassandra/cql3/SystemKeyspaceQueryTest.java @@ -26,7 +26,7 @@ import org.apache.cassandra.utils.TimeUUID; import static org.apache.cassandra.db.SystemKeyspace.BATCHES; import static org.apache.cassandra.db.SystemKeyspace.LOCAL; import static org.apache.cassandra.db.SystemKeyspace.METADATA_LOG; -import static org.apache.cassandra.db.SystemKeyspace.SNAPSHOT_TABLE_NAME; +import static org.apache.cassandra.db.SystemKeyspace.METADATA_SNAPSHOTS; import static org.apache.cassandra.schema.SchemaConstants.SYSTEM_KEYSPACE_NAME; public class SystemKeyspaceQueryTest extends CQLTester @@ -44,7 +44,7 @@ public class SystemKeyspaceQueryTest extends CQLTester assertRowCountNet(executeNet(String.format("SELECT * FROM %s.%s WHERE epoch = 1", SYSTEM_KEYSPACE_NAME, METADATA_LOG)), 0); assertRowCountNet(executeNet(String.format("SELECT * FROM %s.%s WHERE epoch = 1", - SYSTEM_KEYSPACE_NAME, SNAPSHOT_TABLE_NAME)), 0); + SYSTEM_KEYSPACE_NAME, METADATA_SNAPSHOTS)), 0); // system.batches table uses LocalPartitioner assertRowCountNet(executeNet(String.format("SELECT * FROM %s.%s WHERE id = %s", SYSTEM_KEYSPACE_NAME, BATCHES, diff --git a/test/unit/org/apache/cassandra/db/CoordinatorLogOffsetsLifecycleTest.java b/test/unit/org/apache/cassandra/db/CoordinatorLogOffsetsLifecycleTest.java index 98472ad1e3..20d098fcb1 100644 --- a/test/unit/org/apache/cassandra/db/CoordinatorLogOffsetsLifecycleTest.java +++ b/test/unit/org/apache/cassandra/db/CoordinatorLogOffsetsLifecycleTest.java @@ -141,14 +141,17 @@ public class CoordinatorLogOffsetsLifecycleTest { String ks = nextKeyspaceName(); String tbl = "tbl"; - TableMetadata tableMetadata = TableMetadata.builder(ks, tbl) - .addPartitionKeyColumn("k", Int32Type.instance) - .addRegularColumn("v", Int32Type.instance) - .build(); + TableParams tableParams = + TableParams.builder() + .memtable(MemtableParams.get(memtableConfig)) + .build(); - TableParams tableParams = tableMetadata.params; - tableParams = tableParams.unbuild().memtable(MemtableParams.get(memtableConfig)).build(); - tableMetadata = tableMetadata.withSwapped(tableParams); + TableMetadata tableMetadata = + TableMetadata.builder(ks, tbl) + .addPartitionKeyColumn("k", Int32Type.instance) + .addRegularColumn("v", Int32Type.instance) + .params(tableParams) + .build(); SchemaLoader.createKeyspace(ks, KeyspaceParams.simple(1, ReplicationType.tracked), tableMetadata); @@ -183,6 +186,7 @@ public class CoordinatorLogOffsetsLifecycleTest // flush 1 { + MutationTrackingService.instance.persistLogStateForTesting(); cfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS); View view = cfs.getTracker().getView(); @@ -216,6 +220,7 @@ public class CoordinatorLogOffsetsLifecycleTest // flush 2 { + MutationTrackingService.instance.persistLogStateForTesting(); cfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS); View view = cfs.getTracker().getView(); diff --git a/test/unit/org/apache/cassandra/replication/CoordinatorLogOffsetsTest.java b/test/unit/org/apache/cassandra/replication/CoordinatorLogOffsetsTest.java index 91a2788ddf..253d81526b 100644 --- a/test/unit/org/apache/cassandra/replication/CoordinatorLogOffsetsTest.java +++ b/test/unit/org/apache/cassandra/replication/CoordinatorLogOffsetsTest.java @@ -20,8 +20,11 @@ package org.apache.cassandra.replication; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.Mutation; +import org.apache.cassandra.db.commitlog.CommitLog; import org.apache.cassandra.db.marshal.Int32Type; import org.apache.cassandra.dht.Murmur3Partitioner; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.Token; import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper; import org.apache.cassandra.distributed.test.tracking.MutationTrackingUtils; import org.apache.cassandra.locator.InetAddressAndPort; @@ -49,7 +52,10 @@ import org.apache.cassandra.net.MessagingService; import org.assertj.core.api.Assertions; import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -263,6 +269,7 @@ public class CoordinatorLogOffsetsTest ClusterMetadataTestHelper.createKeyspace(ks, KeyspaceParams.simple(3, ReplicationType.tracked)); ClusterMetadataTestHelper.commit(new AlterSchema(SchemaTransformations.addTable(tableMetadata, false))); + CommitLog.instance.start(); MutationTrackingService.instance.start(metadata); // Eventually, will also run perturbations before checking isReconciled (like log truncation, durability, etc.) @@ -276,11 +283,17 @@ public class CoordinatorLogOffsetsTest MutationTrackingService.instance.finishWriting(mutation); MutationTrackingService.instance.receivedWriteResponse(mutation.id(), addr2); MutationTrackingService.instance.receivedWriteResponse(mutation.id(), addr3); + MutationTrackingService.instance.persistLogStateForTesting(); ImmutableCoordinatorLogOffsets logOffsets = new ImmutableCoordinatorLogOffsets.Builder() .add(mutation.id()) .build(); - Assertions.assertThat(MutationTrackingService.instance.isDurablyReconciled(ks, logOffsets)).isTrue(); + Range range = getShardRange(mutation); + List offsets = Collections.singletonList(logOffsets.offsets(mutation.id().logId())); + MutationTrackingService.instance.updateReplicatedOffsets(ks, range, offsets, true, addr2); + MutationTrackingService.instance.updateReplicatedOffsets(ks, range, offsets, true, addr3); + + Assertions.assertThat(MutationTrackingService.instance.isDurablyReconciled(logOffsets)).isTrue(); } // Applied locally but not on remote replicas @@ -288,11 +301,12 @@ public class CoordinatorLogOffsetsTest Mutation mutation = MutationTrackingUtils.createMutation(tableMetadata, 2, 2); MutationTrackingService.instance.startWriting(mutation); MutationTrackingService.instance.finishWriting(mutation); + MutationTrackingService.instance.persistLogStateForTesting(); ImmutableCoordinatorLogOffsets logOffsets = new ImmutableCoordinatorLogOffsets.Builder() .add(mutation.id()) .build(); - Assertions.assertThat(MutationTrackingService.instance.isDurablyReconciled(ks, logOffsets)).isFalse(); + Assertions.assertThat(MutationTrackingService.instance.isDurablyReconciled(logOffsets)).isFalse(); } // Applied on remote replicas but not locally @@ -302,11 +316,18 @@ public class CoordinatorLogOffsetsTest MutationTrackingService.instance.receivedWriteResponse(mutation.id(), addr2); MutationTrackingService.instance.receivedWriteResponse(mutation.id(), addr3); + MutationTrackingService.instance.persistLogStateForTesting(); ImmutableCoordinatorLogOffsets logOffsets = new ImmutableCoordinatorLogOffsets.Builder() .add(mutation.id()) .build(); - Assertions.assertThat(MutationTrackingService.instance.isDurablyReconciled(ks, logOffsets)).isFalse(); + + Range range = getShardRange(mutation); + List offsets = Collections.singletonList(logOffsets.offsets(mutation.id().logId())); + MutationTrackingService.instance.updateReplicatedOffsets(ks, range, offsets, true, addr2); + MutationTrackingService.instance.updateReplicatedOffsets(ks, range, offsets, true, addr3); + + Assertions.assertThat(MutationTrackingService.instance.isDurablyReconciled(logOffsets)).isFalse(); } // If no replicas are aware of a log, it should be considered unreconciled out of caution @@ -317,6 +338,7 @@ public class CoordinatorLogOffsetsTest MutationTrackingService.instance.finishWriting(mutation); MutationTrackingService.instance.receivedWriteResponse(mutation.id(), addr2); MutationTrackingService.instance.receivedWriteResponse(mutation.id(), addr3); + MutationTrackingService.instance.persistLogStateForTesting(); MutationId fakeMutationId = new MutationId(CoordinatorLogId.asLong(111, 222), MutationId.sequenceId(333, 444)); Assertions.assertThat(metadata.directory.version(new NodeId(fakeMutationId.hostId()))).isNull(); @@ -326,9 +348,22 @@ public class CoordinatorLogOffsetsTest ImmutableCoordinatorLogOffsets.Builder logOffsetsBuilder = new ImmutableCoordinatorLogOffsets.Builder(); logOffsetsBuilder.add(fakeMutationId); - Assertions.assertThat(MutationTrackingService.instance.isDurablyReconciled(ks, logOffsetsBuilder.build())).isFalse(); + Assertions.assertThatThrownBy(() -> MutationTrackingService.instance.isDurablyReconciled(logOffsetsBuilder.build())) + .hasSameClassAs(new IllegalStateException()) + .hasMessageMatching("Could not find shard for logId \\d+"); } MutationTrackingService.instance.shutdownBlocking(); + CommitLog.instance.stopUnsafe(true); + } + + private Range getShardRange(Mutation mutation) + { + Map> ksRanges = new HashMap<>(); + MutationTrackingService.instance.forEachKeyspace(shards -> { + Shard shard = shards.lookUp(mutation); + ksRanges.put(shard.keyspace, shard.range); + }); + return ksRanges.get(mutation.getKeyspaceName()); } } \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/replication/CoordinatorLogTest.java b/test/unit/org/apache/cassandra/replication/CoordinatorLogTest.java index f89f1a521b..5122683b41 100644 --- a/test/unit/org/apache/cassandra/replication/CoordinatorLogTest.java +++ b/test/unit/org/apache/cassandra/replication/CoordinatorLogTest.java @@ -21,7 +21,6 @@ import java.io.IOException; import java.util.ArrayList; import java.util.List; -import org.junit.Assert; import org.junit.BeforeClass; import org.junit.Test; @@ -30,6 +29,7 @@ import org.apache.cassandra.db.Mutation; import org.apache.cassandra.db.RowUpdateBuilder; import org.apache.cassandra.db.marshal.UTF8Type; import org.apache.cassandra.dht.ByteOrderedPartitioner; +import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.replication.CoordinatorLog.CoordinatorLogPrimary; import org.apache.cassandra.schema.KeyspaceParams; @@ -38,11 +38,21 @@ import org.apache.cassandra.schema.TableId; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.utils.ByteBufferUtil; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + public class CoordinatorLogTest { private static final int LOCAL_HOST_ID = 1; - private static final CoordinatorLogId LOG_ID = new CoordinatorLogId(LOCAL_HOST_ID, 1); - private static final Participants PARTICIPANTS = new Participants(List.of(LOCAL_HOST_ID, 2, 3)); + private static final int REMOTE_HOST_ID_1 = 2; + private static final int REMOTE_HOST_ID_2 = 3; + + private static final CoordinatorLogId LOCAL_LOG_ID = new CoordinatorLogId(LOCAL_HOST_ID, 1); + private static final CoordinatorLogId REPLICA_LOG_ID = new CoordinatorLogId(REMOTE_HOST_ID_1, 1); + + private static final Participants PARTICIPANTS = + new Participants(List.of(LOCAL_HOST_ID, REMOTE_HOST_ID_1, REMOTE_HOST_ID_2)); private static final String KEYSPACE = "cltks"; private static final String TABLE = "cltt"; @@ -57,6 +67,7 @@ public class CoordinatorLogTest .addClusteringColumn("ck", UTF8Type.instance) .addRegularColumn("value", UTF8Type.instance) .build()); + MutationJournal.instance.start(); } private static Token tk(String key) @@ -66,7 +77,7 @@ public class CoordinatorLogTest private static Offsets toOffsets(MutationId... ids) { - Offsets.Mutable list = new Offsets.Mutable(LOG_ID); + Offsets.Mutable list = new Offsets.Mutable(LOCAL_LOG_ID); for (MutationId id : ids) list.add(id.offset()); return list; @@ -74,15 +85,15 @@ public class CoordinatorLogTest private static void assertUnreconciled(Token token, TableId tableId, CoordinatorLog log, boolean includePending, Offsets expectedReconciled, MutationId... expectedIds) { - Offsets.Mutable reconciled = new Offsets.Mutable(LOG_ID); - Offsets.Mutable unreconciled = new Offsets.Mutable(LOG_ID); + Offsets.Mutable reconciled = new Offsets.Mutable(LOCAL_LOG_ID); + Offsets.Mutable unreconciled = new Offsets.Mutable(LOCAL_LOG_ID); log.collectOffsetsFor(token, tableId, includePending, unreconciled, reconciled); for (MutationId mid : expectedIds) - Assert.assertTrue(unreconciled.contains(mid.offset())); + assertTrue(unreconciled.contains(mid.offset())); - Assert.assertEquals(toOffsets(expectedIds), unreconciled); - Assert.assertEquals(expectedReconciled, reconciled); + assertEquals(toOffsets(expectedIds), unreconciled); + assertEquals(expectedReconciled, reconciled); } @Test @@ -91,24 +102,18 @@ public class CoordinatorLogTest Token tk = tk("key"); TableMetadata metadata = Schema.instance.getTableMetadata(KEYSPACE, TABLE); TableId tableId = metadata.id; - CoordinatorLogPrimary log = new CoordinatorLogPrimary(LOCAL_HOST_ID, LOG_ID, PARTICIPANTS); + CoordinatorLogPrimary log = new CoordinatorLogPrimary(KEYSPACE, new Range<>(tk, tk), LOCAL_HOST_ID, LOCAL_LOG_ID, PARTICIPANTS); MutationId[] ids = new MutationId[] { log.nextId(), log.nextId(), log.nextId(), }; List mutations = new ArrayList<>(ids.length); for (MutationId id : ids) { - Mutation mutation = - new RowUpdateBuilder(metadata, 0, "key") - .clustering("ck") - .add("value", "value") - .build() - .withMutationId(id); - + Mutation mutation = createMutation(id); mutations.add(mutation); log.startWriting(mutation); } - Offsets.Mutable reconciled = new Offsets.Mutable(LOG_ID); + Offsets.Mutable reconciled = new Offsets.Mutable(LOCAL_LOG_ID); // we've only started writing, so the ids shouldn't appear without includePending being true assertUnreconciled(tk, tableId, log, false, reconciled); assertUnreconciled(tk, tableId, log, true, reconciled, ids); @@ -126,4 +131,86 @@ public class CoordinatorLogTest reconciled.add(ids[0].offset()); assertUnreconciled(tk, tableId, log, false, reconciled, ids[1], ids[2]); } + + private Mutation createMutation(MutationId id) + { + TableMetadata metadata = Schema.instance.getTableMetadata(KEYSPACE, TABLE); + return new RowUpdateBuilder(metadata, 0, "key") + .clustering("ck") + .add("value", "value") + .build() + .withMutationId(id); + } + + @Test + public void persistAndLoadPrimaryLogTest() + { + testPersistAndLoadRoundtrip(LOCAL_LOG_ID); + } + + @Test + public void persistAndLoadReplicaLogTest() + { + testPersistAndLoadRoundtrip(REPLICA_LOG_ID); + } + + private void testPersistAndLoadRoundtrip(CoordinatorLogId logId) + { + Range range = new Range<>(tk("a"), tk("b")); + + Offsets.Mutable offsets1 = new Offsets.Mutable(logId); + offsets1.add(1, 2, 3, 4); + Offsets.Mutable offsets2 = new Offsets.Mutable(logId); + offsets2.add(2, 3, 4, 5); + Offsets.Mutable offsets3 = new Offsets.Mutable(logId); + offsets3.add(3, 4, 5, 6); + + Node2OffsetsMap witnessed = new Node2OffsetsMap(); + witnessed.set(LOCAL_HOST_ID, offsets1); + witnessed.set(REMOTE_HOST_ID_1, offsets2); + witnessed.set(REMOTE_HOST_ID_2, offsets3); + + UnreconciledMutations unreconciled = new UnreconciledMutations(); + Mutation mutation1 = createMutation(new MutationId(logId.asLong(), MutationId.sequenceId(1, 0))); + Mutation mutation2 = createMutation(new MutationId(logId.asLong(), MutationId.sequenceId(2, 0))); + unreconciled.addDirectly(mutation1); + unreconciled.addDirectly(mutation2); + MutationJournal.instance.write(mutation1.id(), mutation1); + MutationJournal.instance.write(mutation2.id(), mutation2); + + CoordinatorLog log = + CoordinatorLog.recreate(KEYSPACE, range, LOCAL_HOST_ID, logId, PARTICIPANTS, witnessed, witnessed, unreconciled); + + Offsets.Mutable reconciled = new Offsets.Mutable(logId); + reconciled.add(3, 4); + assertEquals(reconciled, log.reconciledOffsets); + + validatePersistAndLoadRoundtrip(log); + log.deleteFromSystemTable(); + } + + private static void validatePersistAndLoadRoundtrip(CoordinatorLog log) + { + log.persistToSystemTable(); + List logs = CoordinatorLog.loadFromSystemTable(KEYSPACE, log.range, LOCAL_HOST_ID); + assertEquals(1, logs.size()); + CoordinatorLog loaded = logs.get(0); + + assertSame(log.getClass(), loaded.getClass()); + assertEquals(log.keyspace, loaded.keyspace); + assertEquals(log.range, loaded.range); + assertEquals(log.logId, loaded.logId); + assertEquals(log.participants, loaded.participants); + assertEquals(log.localNodeId, loaded.localNodeId); + + assertEquals(log.participants.size(), log.witnessedOffsets.size()); + assertEquals(log.participants.size(), log.persistedOffsets.size()); + assertEquals(loaded.participants.size(), loaded.witnessedOffsets.size()); + assertEquals(loaded.participants.size(), loaded.persistedOffsets.size()); + assertEquals(log.witnessedOffsets, loaded.witnessedOffsets); + assertEquals(log.persistedOffsets, loaded.persistedOffsets); + assertEquals(log.reconciledOffsets, loaded.reconciledOffsets); + + assertTrue(log.unreconciledMutations.equalsForTesting(loaded.unreconciledMutations)); + } } diff --git a/test/unit/org/apache/cassandra/replication/OffsetsTest.java b/test/unit/org/apache/cassandra/replication/OffsetsTest.java index 61c0a9ba5f..b0d5edda0f 100644 --- a/test/unit/org/apache/cassandra/replication/OffsetsTest.java +++ b/test/unit/org/apache/cassandra/replication/OffsetsTest.java @@ -843,21 +843,27 @@ public class OffsetsTest public void testRemoveWithExactSizedArray() { { - Offsets.Mutable offsets = new Offsets.Mutable(LOG_ID, new int[] {10, 11}, 2); + Offsets.Mutable offsets = new Offsets.Mutable(LOG_ID, new int[]{ 10, 11 }, 2); offsets.remove(10); assertOffsetsEqual(offsets(11, 11), offsets); } { - Offsets.Mutable offsets = new Offsets.Mutable(LOG_ID, new int[] {10, 11}, 2); + Offsets.Mutable offsets = new Offsets.Mutable(LOG_ID, new int[]{ 10, 11 }, 2); offsets.remove(11); assertOffsetsEqual(offsets(10, 10), offsets); } { - Offsets.Mutable offsets = new Offsets.Mutable(LOG_ID, new int[] {11, 11}, 2); + Offsets.Mutable offsets = new Offsets.Mutable(LOG_ID, new int[]{ 11, 11 }, 2); offsets.remove(11); assertOffsetsEqual(offsets(), offsets); } } + + public void asListFromListRoundTripTest() + { + for (Offsets.Mutable offsets : new Offsets.Mutable[] { offsets(), offsets(1, 2), offsets(1, 3, 7, 9) }) + assertOffsetsEqual(offsets, Offsets.fromList(LOG_ID, offsets.asList())); + } } diff --git a/test/unit/org/apache/cassandra/replication/ShardTest.java b/test/unit/org/apache/cassandra/replication/ShardTest.java new file mode 100644 index 0000000000..d8c62554f1 --- /dev/null +++ b/test/unit/org/apache/cassandra/replication/ShardTest.java @@ -0,0 +1,112 @@ +/* + * 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.replication; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.function.LongSupplier; + +import org.junit.BeforeClass; +import org.junit.Test; + +import org.agrona.collections.MutableInteger; +import org.apache.cassandra.SchemaLoader; +import org.apache.cassandra.db.marshal.UTF8Type; +import org.apache.cassandra.dht.ByteOrderedPartitioner; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.schema.KeyspaceParams; +import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.utils.ByteBufferUtil; + +import static org.junit.Assert.assertEquals; + +public class ShardTest +{ + private static final int LOCAL_HOST_ID = 1; + private static final int REMOTE_HOST_ID_1 = 2; + private static final int REMOTE_HOST_ID_2 = 3; + + private static final String KEYSPACE = "shard_test_ks"; + private static final String TABLE = "shard_test_table"; + + @BeforeClass + public static void setUp() throws IOException + { + SchemaLoader.prepareServer(); + SchemaLoader.createKeyspace(KEYSPACE, KeyspaceParams.simple(3), + TableMetadata.builder(KEYSPACE, TABLE) + .addPartitionKeyColumn("pk", UTF8Type.instance) + .addClusteringColumn("ck", UTF8Type.instance) + .addRegularColumn("value", UTF8Type.instance) + .build()); + MutationJournal.instance.start(); + } + + private static Token tk(String key) + { + return new ByteOrderedPartitioner.BytesToken(ByteBufferUtil.bytes(key)); + } + + @Test + public void testPersistAndLoadSingleShard() + { + Range range = new Range<>(tk("a"), tk("z")); + Participants participants = new Participants(List.of(LOCAL_HOST_ID, REMOTE_HOST_ID_1, REMOTE_HOST_ID_2)); + MutableInteger logId = new MutableInteger(); + LongSupplier logIdProvider = () -> CoordinatorLogId.asLong(LOCAL_HOST_ID, logId.getAndIncrement()); + + Shard original = new Shard(LOCAL_HOST_ID, KEYSPACE, range, participants, logIdProvider, (s, l) -> {}); + original.persistToSystemTables(); + + ArrayList loadedShards = Shard.loadFromSystemTables(LOCAL_HOST_ID, logIdProvider, (s, l) -> {}); + assertEquals(1, loadedShards.size()); + Shard loaded = loadedShards.get(0); + + assertEquals(original.localNodeId, loaded.localNodeId); + assertEquals(original.keyspace, loaded.keyspace); + assertEquals(original.range, loaded.range); + assertEquals(original.participants, loaded.participants); + // TODO: compare the coordinator logs + } + + @Test + public void testLogRotation() + { + CoordinatorLog.overrideMaxOffsetForTesting(100); + try + { + Range range = new Range<>(tk("a"), tk("z")); + Participants participants = new Participants(List.of(LOCAL_HOST_ID, REMOTE_HOST_ID_1, REMOTE_HOST_ID_2)); + MutableInteger logId = new MutableInteger(); + LongSupplier logIdProvider = () -> CoordinatorLogId.asLong(LOCAL_HOST_ID, logId.getAndIncrement()); + Shard shard = new Shard(LOCAL_HOST_ID, KEYSPACE, range, participants, logIdProvider, (s, l) -> { + }); + + MutationId firstId = shard.nextId(); + for (int i = 0; i < 100; i++) + assertEquals(firstId.hostLogId, shard.nextId().hostLogId); + assertEquals(firstId.hostLogId + 1, shard.nextId().hostLogId); + } + finally + { + CoordinatorLog.overrideMaxOffsetForTesting(Integer.MAX_VALUE); + } + } +} \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/replication/UnreconciledMutationsTest.java b/test/unit/org/apache/cassandra/replication/UnreconciledMutationsTest.java index dfa025a327..281c8da079 100644 --- a/test/unit/org/apache/cassandra/replication/UnreconciledMutationsTest.java +++ b/test/unit/org/apache/cassandra/replication/UnreconciledMutationsTest.java @@ -43,10 +43,13 @@ import org.apache.cassandra.schema.TableId; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.utils.ByteBufferUtil; +import static org.junit.Assert.assertEquals; + public class UnreconciledMutationsTest { private static final String KEYSPACE = "ks"; private static final String TABLE = "tbl"; + private static TableId TABLE_ID; @BeforeClass public static void setUp() throws IOException @@ -57,6 +60,7 @@ public class UnreconciledMutationsTest .addPartitionKeyColumn("k", Int32Type.instance) .addRegularColumn("v", Int32Type.instance) .build()); + TABLE_ID = Schema.instance.getTableMetadata(KEYSPACE, TABLE).id; } private static Token tokenFor(int key) @@ -69,9 +73,7 @@ public class UnreconciledMutationsTest TableMetadata metadata = Schema.instance.getTableMetadata(KEYSPACE, TABLE); // Create a MutationId with logId 1L and sequenceId that produces the desired offset - long logId = 1L; - long sequenceId = ((long) offset << 32) | 1000; // offset in high bits, timestamp in low bits - MutationId mutationId = new MutationId(logId, sequenceId); + MutationId mutationId = new MutationId(CoordinatorLogId.asLong(1, 1), MutationId.sequenceId(offset, 0)); Mutation mutation = new RowUpdateBuilder(metadata, 0, partitionKey) .add("v", value) @@ -90,8 +92,7 @@ public class UnreconciledMutationsTest public void testSingleTokenCollectionIsolation() { UnreconciledMutations unreconciled = new UnreconciledMutations(); - TableId tableId = Schema.instance.getTableMetadata(KEYSPACE, TABLE).id; - + // Create mutations for different partition keys with different tokens Mutation mutation1 = createMutation(100, 1000, 1); Mutation mutation2 = createMutation(200, 2000, 2); @@ -113,30 +114,30 @@ public class UnreconciledMutationsTest // Test single token collection for token1 - should ONLY return mutation1 Offsets.Mutable offsets1 = new Offsets.Mutable(new CoordinatorLogId(1, 1)); - boolean found1 = unreconciled.collect(token1, tableId, false, offsets1); + boolean found1 = unreconciled.collect(token1, TABLE_ID, false, offsets1); Assert.assertTrue("Should find mutations for token1", found1); - Assert.assertEquals("Should only contain 1 offset for token1", 1, offsets1.offsetCount()); + assertEquals("Should only contain 1 offset for token1", 1, offsets1.offsetCount()); Assert.assertTrue("Should contain offset 1 for mutation1", offsets1.contains(1)); Assert.assertFalse("Should NOT contain offset 2 for mutation2", offsets1.contains(2)); Assert.assertFalse("Should NOT contain offset 3 for mutation3", offsets1.contains(3)); // Test single token collection for token2 - should ONLY return mutation2 Offsets.Mutable offsets2 = new Offsets.Mutable(new CoordinatorLogId(1, 1)); - boolean found2 = unreconciled.collect(token2, tableId, false, offsets2); + boolean found2 = unreconciled.collect(token2, TABLE_ID, false, offsets2); Assert.assertTrue("Should find mutations for token2", found2); - Assert.assertEquals("Should only contain 1 offset for token2", 1, offsets2.offsetCount()); + assertEquals("Should only contain 1 offset for token2", 1, offsets2.offsetCount()); Assert.assertTrue("Should contain offset 2 for mutation2", offsets2.contains(2)); Assert.assertFalse("Should NOT contain offset 1 for mutation1", offsets2.contains(1)); Assert.assertFalse("Should NOT contain offset 3 for mutation3", offsets2.contains(3)); // Test single token collection for token3 - should ONLY return mutation3 Offsets.Mutable offsets3 = new Offsets.Mutable(new CoordinatorLogId(1, 1)); - boolean found3 = unreconciled.collect(token3, tableId, false, offsets3); + boolean found3 = unreconciled.collect(token3, TABLE_ID, false, offsets3); Assert.assertTrue("Should find mutations for token3", found3); - Assert.assertEquals("Should only contain 1 offset for token3", 1, offsets3.offsetCount()); + assertEquals("Should only contain 1 offset for token3", 1, offsets3.offsetCount()); Assert.assertTrue("Should contain offset 3 for mutation3", offsets3.contains(3)); Assert.assertFalse("Should NOT contain offset 1 for mutation1", offsets3.contains(1)); Assert.assertFalse("Should NOT contain offset 2 for mutation2", offsets3.contains(2)); @@ -146,11 +147,10 @@ public class UnreconciledMutationsTest public void testEmptyCollection() { UnreconciledMutations unreconciled = new UnreconciledMutations(); - TableId tableId = Schema.instance.getTableMetadata(KEYSPACE, TABLE).id; Token token = tokenFor(100); Offsets.Mutable offsets = new Offsets.Mutable(new CoordinatorLogId(1, 1)); - boolean found = unreconciled.collect(token, tableId, false, offsets); + boolean found = unreconciled.collect(token, TABLE_ID, false, offsets); Assert.assertFalse("Should not find any mutations in empty collection", found); Assert.assertTrue("Should have no offsets", offsets.isEmpty()); @@ -160,8 +160,7 @@ public class UnreconciledMutationsTest public void testPendingVsVisibleMutations() { UnreconciledMutations unreconciled = new UnreconciledMutations(); - TableId tableId = Schema.instance.getTableMetadata(KEYSPACE, TABLE).id; - + Mutation pendingMutation = createMutation(100, 1000, 1); Mutation visibleMutation = createMutation(100, 2000, 2); Token token = pendingMutation.key().getToken(); @@ -173,19 +172,19 @@ public class UnreconciledMutationsTest // Test without including pending - should only get visible mutation Offsets.Mutable visibleOnly = new Offsets.Mutable(new CoordinatorLogId(1, 1)); - boolean foundVisible = unreconciled.collect(token, tableId, false, visibleOnly); + boolean foundVisible = unreconciled.collect(token, TABLE_ID, false, visibleOnly); Assert.assertTrue("Should find visible mutations", foundVisible); - Assert.assertEquals("Should only have 1 visible mutation", 1, visibleOnly.offsetCount()); + assertEquals("Should only have 1 visible mutation", 1, visibleOnly.offsetCount()); Assert.assertTrue("Should contain visible mutation offset", visibleOnly.contains(2)); Assert.assertFalse("Should NOT contain pending mutation offset", visibleOnly.contains(1)); // Test including pending - should get both mutations Offsets.Mutable includingPending = new Offsets.Mutable(new CoordinatorLogId(1, 1)); - boolean foundAll = unreconciled.collect(token, tableId, true, includingPending); + boolean foundAll = unreconciled.collect(token, TABLE_ID, true, includingPending); Assert.assertTrue("Should find all mutations", foundAll); - Assert.assertEquals("Should have 2 mutations total", 2, includingPending.offsetCount()); + assertEquals("Should have 2 mutations total", 2, includingPending.offsetCount()); Assert.assertTrue("Should contain pending mutation offset", includingPending.contains(1)); Assert.assertTrue("Should contain visible mutation offset", includingPending.contains(2)); } @@ -194,8 +193,7 @@ public class UnreconciledMutationsTest public void testMultipleMutationsSameToken() { UnreconciledMutations unreconciled = new UnreconciledMutations(); - TableId tableId = Schema.instance.getTableMetadata(KEYSPACE, TABLE).id; - + // Create multiple mutations for the same partition key (same token) Mutation mutation1 = createMutation(100, 1000, 1); Mutation mutation2 = createMutation(100, 2000, 2); @@ -211,10 +209,10 @@ public class UnreconciledMutationsTest unreconciled.finishWriting(mutation3); Offsets.Mutable offsets = new Offsets.Mutable(new CoordinatorLogId(1, 1)); - boolean found = unreconciled.collect(token, tableId, false, offsets); + boolean found = unreconciled.collect(token, TABLE_ID, false, offsets); Assert.assertTrue("Should find mutations for token", found); - Assert.assertEquals("Should have all 3 mutations for same token", 3, offsets.offsetCount()); + assertEquals("Should have all 3 mutations for same token", 3, offsets.offsetCount()); Assert.assertTrue("Should contain offset 1", offsets.contains(1)); Assert.assertTrue("Should contain offset 2", offsets.contains(2)); Assert.assertTrue("Should contain offset 3", offsets.contains(3)); @@ -224,8 +222,7 @@ public class UnreconciledMutationsTest public void testTableIdFiltering() { UnreconciledMutations unreconciled = new UnreconciledMutations(); - TableId tableId = Schema.instance.getTableMetadata(KEYSPACE, TABLE).id; - + // Create a fake different table ID TableId differentTableId = TableId.generate(); @@ -237,10 +234,10 @@ public class UnreconciledMutationsTest // Query with correct table ID - should find mutation Offsets.Mutable correctTable = new Offsets.Mutable(new CoordinatorLogId(1, 1)); - boolean foundCorrect = unreconciled.collect(token, tableId, false, correctTable); + boolean foundCorrect = unreconciled.collect(token, TABLE_ID, false, correctTable); Assert.assertTrue("Should find mutation for correct table", foundCorrect); - Assert.assertEquals("Should have 1 mutation", 1, correctTable.offsetCount()); + assertEquals("Should have 1 mutation", 1, correctTable.offsetCount()); // Query with different table ID - should find nothing Offsets.Mutable differentTable = new Offsets.Mutable(new CoordinatorLogId(1, 1)); @@ -254,8 +251,7 @@ public class UnreconciledMutationsTest public void testMutationRemoval() { UnreconciledMutations unreconciled = new UnreconciledMutations(); - TableId tableId = Schema.instance.getTableMetadata(KEYSPACE, TABLE).id; - + Mutation mutation = createMutation(100, 1000, 1); Token token = mutation.key().getToken(); @@ -264,16 +260,16 @@ public class UnreconciledMutationsTest // Verify mutation is present Offsets.Mutable beforeRemoval = new Offsets.Mutable(new CoordinatorLogId(1, 1)); - boolean foundBefore = unreconciled.collect(token, tableId, false, beforeRemoval); + boolean foundBefore = unreconciled.collect(token, TABLE_ID, false, beforeRemoval); Assert.assertTrue("Should find mutation before removal", foundBefore); - Assert.assertEquals("Should have 1 mutation before removal", 1, beforeRemoval.offsetCount()); + assertEquals("Should have 1 mutation before removal", 1, beforeRemoval.offsetCount()); // Remove the mutation unreconciled.remove(1); // Verify mutation is gone Offsets.Mutable afterRemoval = new Offsets.Mutable(new CoordinatorLogId(1, 1)); - boolean foundAfter = unreconciled.collect(token, tableId, false, afterRemoval); + boolean foundAfter = unreconciled.collect(token, TABLE_ID, false, afterRemoval); Assert.assertFalse("Should not find mutation after removal", foundAfter); Assert.assertTrue("Should have no mutations after removal", afterRemoval.isEmpty()); } @@ -282,8 +278,7 @@ public class UnreconciledMutationsTest public void testTokenRangeCollection() { UnreconciledMutations unreconciled = new UnreconciledMutations(); - TableId tableId = Schema.instance.getTableMetadata(KEYSPACE, TABLE).id; - + // Create mutations with different tokens and sort them List keys = List.of(100, 200, 300, 400, 500); List mutations = new ArrayList<>(); @@ -311,7 +306,7 @@ public class UnreconciledMutationsTest new Bounds<>(startKey, endKey); Offsets.Mutable rangeOffsets = new Offsets.Mutable(new CoordinatorLogId(1, 1)); - boolean foundRange = unreconciled.collect(bounds, tableId, false, rangeOffsets); + boolean foundRange = unreconciled.collect(bounds, TABLE_ID, false, rangeOffsets); Assert.assertTrue("Should find mutations in range", foundRange); // Should include mutations at positions 0, 1, and 2 (first through middle inclusive) @@ -322,8 +317,7 @@ public class UnreconciledMutationsTest public void testSingleTokenRangeCollectionBug() { UnreconciledMutations unreconciled = new UnreconciledMutations(); - TableId tableId = Schema.instance.getTableMetadata(KEYSPACE, TABLE).id; - + // Create mutations for different partition keys with different tokens Mutation mutation1 = createMutation(100, 1000, 1); Mutation mutation2 = createMutation(200, 2000, 2); @@ -342,11 +336,11 @@ public class UnreconciledMutationsTest // Test single token range collection Offsets.Mutable offsets = new Offsets.Mutable(new CoordinatorLogId(1, 1)); Bounds bounds = new Bounds<>(mutation1.key(), mutation1.key()); - boolean found = unreconciled.collect(bounds, tableId, false, offsets); + boolean found = unreconciled.collect(bounds, TABLE_ID, false, offsets); // This should only return mutation1 Assert.assertTrue("Should find mutations for single token range", found); - Assert.assertEquals("Single token range should only return 1 mutation", 1, offsets.offsetCount()); + assertEquals("Single token range should only return 1 mutation", 1, offsets.offsetCount()); Assert.assertTrue("Should contain mutation1 offset", offsets.contains(1)); // other mutations should not be included @@ -358,8 +352,7 @@ public class UnreconciledMutationsTest public void testFullRangeCollectionWithMinimumToken() { UnreconciledMutations unreconciled = new UnreconciledMutations(); - TableId tableId = Schema.instance.getTableMetadata(KEYSPACE, TABLE).id; - + // Create mutations for different partition keys with different tokens Mutation mutation1 = createMutation(100, 1000, 1); Mutation mutation2 = createMutation(200, 2000, 2); @@ -382,11 +375,11 @@ public class UnreconciledMutationsTest // Test full range collection - this SHOULD return ALL mutations Offsets.Mutable offsets = new Offsets.Mutable(new CoordinatorLogId(1, 1)); - boolean found = unreconciled.collect(fullRange, tableId, false, offsets); + boolean found = unreconciled.collect(fullRange, TABLE_ID, false, offsets); // Full range should return ALL mutations Assert.assertTrue("Should find mutations for full range", found); - Assert.assertEquals("Full range should return ALL mutations", 3, offsets.offsetCount()); + assertEquals("Full range should return ALL mutations", 3, offsets.offsetCount()); Assert.assertTrue("Should contain mutation1 offset", offsets.contains(1)); Assert.assertTrue("Should contain mutation2 offset", offsets.contains(2)); Assert.assertTrue("Should contain mutation3 offset", offsets.contains(3)); @@ -396,8 +389,7 @@ public class UnreconciledMutationsTest public void testSingleTokenCollectionVsFullRange() { UnreconciledMutations unreconciled = new UnreconciledMutations(); - TableId tableId = Schema.instance.getTableMetadata(KEYSPACE, TABLE).id; - + // Create mutations with different tokens Mutation mutation1 = createMutation(100, 1000, 1); Mutation mutation2 = createMutation(200, 2000, 2); @@ -415,10 +407,10 @@ public class UnreconciledMutationsTest // Single token collection should only return mutation1 Offsets.Mutable singleToken = new Offsets.Mutable(new CoordinatorLogId(1, 1)); - boolean foundSingle = unreconciled.collect(targetToken, tableId, false, singleToken); + boolean foundSingle = unreconciled.collect(targetToken, TABLE_ID, false, singleToken); Assert.assertTrue("Should find mutation for single token", foundSingle); - Assert.assertEquals("Single token should only return 1 mutation", 1, singleToken.offsetCount()); + assertEquals("Single token should only return 1 mutation", 1, singleToken.offsetCount()); Assert.assertTrue("Should contain mutation1 offset", singleToken.contains(1)); // This is the CRITICAL test - single token collection must NOT return all mutations @@ -426,4 +418,33 @@ public class UnreconciledMutationsTest Assert.assertFalse("Single token collection should NOT contain mutation2", singleToken.contains(2)); Assert.assertFalse("Single token collection should NOT contain mutation3", singleToken.contains(3)); } + + @Test + public void testLoadFromJournal() + { + MutationJournal.instance.start(); + + CoordinatorLogId logId = new CoordinatorLogId(1, 1); + + Offsets.Mutable offsets1 = new Offsets.Mutable(logId); + offsets1.add(1, 2, 3, 4, 5, 6, 7); + + Offsets.Mutable offsets2 = new Offsets.Mutable(logId); + offsets2.add(1, 2, 3, 4, 5); + + Node2OffsetsMap witnessed = new Node2OffsetsMap(); + witnessed.set(1, offsets1); + witnessed.set(2, offsets2); + + Mutation mutation6 = createMutation(6, 6, 6); + Mutation mutation7 = createMutation(7, 7, 7); + MutationJournal.instance.write(mutation6.id(), mutation6); + MutationJournal.instance.write(mutation7.id(), mutation7); + + Offsets.Mutable loadedOffsets = new Offsets.Mutable(logId); + UnreconciledMutations unreconciled = UnreconciledMutations.loadFromJournal(witnessed, 1); + unreconciled.collect(mutation6.key().getToken(), TABLE_ID, false, loadedOffsets); + unreconciled.collect(mutation7.key().getToken(), TABLE_ID, false, loadedOffsets); + assertEquals(List.of(6, 7), loadedOffsets.asList()); + } } \ No newline at end of file