natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Implement Shard and CoordinatorLog metadata durability 

patch by Aleksey Yeschenko; reviewed by Abe Ratnofsky for CASSANDRA-20882
This commit is contained in:
Aleksey Yeschenko 2025-07-17 13:00:33 +01:00 committed by Blake Eggleston
parent 9dbafa812d
commit 9d856f2c08
19 changed files with 1102 additions and 341 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

80
src/java/org/apache/cassandra/db/SystemKeyspace.java
View File

@ -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<String> 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<uuid>, "
+ "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<set<int>>,"
+ "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<set<int>>,"
+ "witnessed_offsets map<int, frozen<list<int>>>,"
+ "persisted_offsets map<int, frozen<list<int>>>,"
+ "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<TableId, Pair<CommitLogPosition, Long>> 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();

2
src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java
View File

@ -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);

12
src/java/org/apache/cassandra/replication/BroadcastLogOffsets.java
View File

@ -41,12 +41,14 @@ public class BroadcastLogOffsets
private final String keyspace;
private final Range<Token> range;
private final List<Offsets.Immutable> replicatedOffsets;
private final boolean durable;
public BroadcastLogOffsets(String keyspace, Range<Token> range, List<Offsets.Immutable> offsets)
public BroadcastLogOffsets(String keyspace, Range<Token> range, List<Offsets.Immutable> 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<BroadcastLogOffsets> 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<Offsets.Immutable> 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;
}
};

308
src/java/org/apache/cassandra/replication/CoordinatorLog.java
View File

@ -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<Token> 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<Token> 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<Token> 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<Token> 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<Token> 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<Token> 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<Token> 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<Token> 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<Token> 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<Integer, List<Integer>> witnessed = new Int2ObjectHashMap<>();
Map<Integer, List<Integer>> 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<Integer, List<Integer>> witnessed = new Int2ObjectHashMap<>();
Map<Integer, List<Integer>> 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<CoordinatorLog> loadFromSystemTable(String keyspace, Range<Token> range, int localNodeId)
{
ArrayList<CoordinatorLog> 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<Integer> participants = row.getFrozenSet("participants", Int32Type.instance);
Map<Integer, List<Integer>> witnessedOffsets =
row.getMap("witnessed_offsets", Int32Type.instance, ListType.getInstance(Int32Type.instance, false));
Map<Integer, List<Integer>> 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;
}

237
src/java/org/apache/cassandra/replication/MutationTrackingService.java
View File

@ -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<CoordinatorLogId, Shard> 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<Token> range, List<? extends Offsets> offsets, InetAddressAndPort onHost)
public void updateReplicatedOffsets(String keyspace, Range<Token> range, List<? extends Offsets> 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<Range<Token>, Shard> shards;
private final ReplicaGroups groups;
private final BiConsumer<Shard, CoordinatorLog> onNewLog;
private transient final Map<Range<PartitionPosition>, Shard> ppShards;
private transient final Map<CoordinatorLogId, CoordinatorLog> logs;
static KeyspaceShards make(KeyspaceMetadata keyspace, ClusterMetadata cluster, IntSupplier logIdProvider, BiConsumer<Shard, CoordinatorLog> onNewLog)
static KeyspaceShards make(KeyspaceMetadata keyspace, ClusterMetadata cluster, LongSupplier logIdProvider, BiConsumer<Shard, CoordinatorLog> onNewLog)
{
Preconditions.checkArgument(keyspace.params.replicationType.isTracked());
@ -341,26 +351,23 @@ public class MutationTrackingService
for (Range<Token> 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<Range<Token>, Shard> shards, ReplicaGroups groups, BiConsumer<Shard, CoordinatorLog> onNewLog)
KeyspaceShards(String keyspace, Map<Range<Token>, Shard> shards, ReplicaGroups groups)
{
this.keyspace = keyspace;
this.shards = shards;
this.groups = groups;
this.onNewLog = onNewLog;
this.logs = new HashMap<>();
HashMap<Range<PartitionPosition>, 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<Token> range, List<? extends Offsets> offsets, InetAddressAndPort onHost)
void updateReplicatedOffsets(Range<Token> range, List<? extends Offsets> 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<KeyspaceShards> loadFromSystemTables(ClusterMetadata cluster, LongSupplier logIdProvider, BiConsumer<Shard, CoordinatorLog> onNewLog)
{
logger.debug("Indexing current log {}", currentLog);
logs.put(currentLog.logId, currentLog);
Map<String, Map<Range<Token>, 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> keyspaceShards = new ArrayList<>();
for (Map.Entry<String, Map<Range<Token>, Shard>> entry : groupedShards.entrySet())
{
ReplicationParams params = cluster.schema.getKeyspaceMetadata(entry.getKey()).params.replication;
ReplicaGroups originalGroups = cluster.placements.get(params).writes; // prior to splitting
Map<Range<Token>, VersionedEndpoints.ForRange> splitGroups = new HashMap<>();
for (Range<Token> 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<BroadcastLogOffsets> 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<BroadcastLogOffsets> 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

89
src/java/org/apache/cassandra/replication/Node2OffsetsMap.java
View File

@ -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<Offsets> map = new Int2ObjectHashMap<>();
private final Int2ObjectHashMap<Offsets.Mutable> 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<Offsets.Mutable> 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<Offsets> consumer)
public void forEach(IntObjConsumer<Offsets.Mutable> consumer)
{
map.forEachInt(consumer);
offsetsMap.forEachInt(consumer);
}
public void clear()
{
map.clear();
offsetsMap.clear();
}
public int size()
{
return offsetsMap.size();
}
void convertToPrimitiveMap(Map<Integer, List<Integer>> into)
{
for (Int2ObjectHashMap<Offsets.Mutable>.EntryIterator iter = offsetsMap.entrySet().iterator(); iter.hasNext();)
{
iter.next();
into.put(iter.getIntKey(), iter.getValue().asList());
}
}
static Node2OffsetsMap fromPrimitiveMap(CoordinatorLogId logId, Map<Integer, List<Integer>> from)
{
Node2OffsetsMap map = new Node2OffsetsMap();
for (Map.Entry<Integer, List<Integer>> 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);
}
}

33
src/java/org/apache/cassandra/replication/Offsets.java
View File

@ -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<ShortMutationId>
{
@ -200,6 +202,19 @@ public abstract class Offsets implements Iterable<ShortMutationId>
digest.updateWithInt(bounds[i]);
}
List<Integer> asList()
{
return new IntArrayList(Arrays.copyOf(bounds, size), size, IntArrayList.DEFAULT_NULL_VALUE);
}
static Mutable fromList(CoordinatorLogId logId, List<Integer> 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<T extends AbstractMutable<T>> extends Offsets implements OffsetReciever
{
AbstractMutable(RangeIterator rangeIterator)
@ -380,6 +395,11 @@ public abstract class Offsets implements Iterable<ShortMutationId>
// 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<ShortMutationId>
{
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

20
src/java/org/apache/cassandra/replication/Participants.java
View File

@ -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<Integer> asSet()
{
IntHashSet set = new IntHashSet(hosts.length);
for (int host : hosts)
set.add(host);
return set;
}
}

188
src/java/org/apache/cassandra/replication/Shard.java
View File

@ -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<Token> 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<Token> range;
final Participants participants;
private final LongSupplier logIdProvider;
private final BiConsumer<Shard, CoordinatorLog> onNewLog;
private final NonBlockingHashMapLong<CoordinatorLog> logs;
// TODO (expected): add support for log rotation
private final CoordinatorLog.CoordinatorLogPrimary currentLocalLog;
private volatile CoordinatorLogPrimary currentLocalLog;
private final List<Subscriber> subscribers = new ArrayList<>();
public interface Subscriber
{
default void onLogCreation(CoordinatorLog log) {}
default void onSubscribe(CoordinatorLog currentLog) {}
}
Shard(String keyspace,
Range<Token> tokenRange,
int localHostId,
Shard(int localNodeId,
String keyspace,
Range<Token> range,
Participants participants,
Epoch sinceEpoch,
IntSupplier logIdProvider,
List<CoordinatorLog> logs,
LongSupplier logIdProvider,
BiConsumer<Shard, CoordinatorLog> 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<Token> range, Participants participants, LongSupplier logIdProvider, BiConsumer<Shard, CoordinatorLog> 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<? extends Offsets> offsets, InetAddressAndPort onHost)
void updateReplicatedOffsets(List<? extends Offsets> 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.Immutable> 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<Shard> loadFromSystemTables(int localNodeId, LongSupplier logIdProvider, BiConsumer<Shard, CoordinatorLog> onNewLog)
{
Token.TokenFactory factory = ClusterMetadata.current().partitioner.getTokenFactory();
ArrayList<Shard> 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<Token> range = new Range<>(factory.fromString(rangeStart), factory.fromString(rangeEnd));
Set<Integer> participants = row.getFrozenSet("participants", Int32Type.instance);
List<CoordinatorLog> logs = CoordinatorLog.loadFromSystemTable(keyspace, range, localNodeId);
shards.add(new Shard(localNodeId, keyspace, range, new Participants(participants), logs, logIdProvider, onNewLog));
}
return shards;
}
}

38
src/java/org/apache/cassandra/replication/UnreconciledMutations.java
View File

@ -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;
}
}

4
src/java/org/apache/cassandra/tools/OfflineClusterMetadataDump.java
View File

@ -285,10 +285,10 @@ public class OfflineClusterMetadataDump implements Runnable
}
// Find and import SSTables for metadata_snapshots
Set<String> snapshotTablePaths = findTablePaths(SystemKeyspace.SNAPSHOT_TABLE_NAME, SchemaConstants.SYSTEM_KEYSPACE_NAME);
Set<String> 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)
{

4
test/distributed/org/apache/cassandra/distributed/test/log/SystemKeyspaceStorageTest.java
View File

@ -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);
}
}

4
test/unit/org/apache/cassandra/cql3/SystemKeyspaceQueryTest.java
View File

@ -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,

19
test/unit/org/apache/cassandra/db/CoordinatorLogOffsetsLifecycleTest.java
View File

@ -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();

43
test/unit/org/apache/cassandra/replication/CoordinatorLogOffsetsTest.java
View File

@ -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<Token> range = getShardRange(mutation);
List<? extends Offsets> 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<Token> range = getShardRange(mutation);
List<? extends Offsets> 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<Token> getShardRange(Mutation mutation)
{
Map<String, Range<Token>> ksRanges = new HashMap<>();
MutationTrackingService.instance.forEachKeyspace(shards -> {
Shard shard = shards.lookUp(mutation);
ksRanges.put(shard.keyspace, shard.range);
});
return ksRanges.get(mutation.getKeyspaceName());
}
}

123
test/unit/org/apache/cassandra/replication/CoordinatorLogTest.java
View File

@ -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<Mutation> 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<Token> 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<CoordinatorLog> 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));
}
}

12
test/unit/org/apache/cassandra/replication/OffsetsTest.java
View File

@ -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()));
}
}

112
test/unit/org/apache/cassandra/replication/ShardTest.java Normal file
View File

@ -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<Token> 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<Shard> 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<Token> 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);
}
}
}

115
test/unit/org/apache/cassandra/replication/UnreconciledMutationsTest.java
View File

@ -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<Integer> keys = List.of(100, 200, 300, 400, 500);
List<Mutation> 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<PartitionPosition> 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());
}
}