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

Fix MutationId class hierarchy and related maps lookup issues 

patch by Aleksey Yeschenko; reviewed by Ariel Weisberg for
CASSANDRA-21366
This commit is contained in:
Aleksey Yeshchenko 2026-05-12 13:44:45 +01:00 committed by Aleksey Yeschenko
parent a79d34a2c0
commit e982cf69b4
8 changed files with 95 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
src/java/org/apache/cassandra/replication/CoordinatorLogId.java
View File

@ -30,17 +30,21 @@ import org.apache.cassandra.io.util.DataOutputPlus;
public class CoordinatorLogId implements Serializable
{
private static final CoordinatorLogId NONE = new CoordinatorLogId(Integer.MIN_VALUE, Integer.MIN_VALUE);
static final int NONE_HOST_ID = Integer.MIN_VALUE;
static final int NONE_HOST_LOG_ID = Integer.MIN_VALUE;
static final long NONE_LOG_ID = asLong(NONE_HOST_ID, NONE_HOST_LOG_ID);
static final CoordinatorLogId NONE = new CoordinatorLogId(NONE_LOG_ID);
/** TCM host ID */
protected final int hostId;
public final int hostId;
/**
* Host log ID (unique within the host).
* Allocated anew on host restart - one per token range replicated by the host.
* Persisted on allocation, unique within the host.
*/
protected final int hostLogId;
public final int hostLogId;
CoordinatorLogId(long id)
{
@ -84,11 +88,6 @@ public class CoordinatorLogId implements Serializable
return (int) coordinatorLogId;
}
public static CoordinatorLogId none()
{
return NONE;
}
public static CoordinatorLogId fromLong(long logId)
{
return new CoordinatorLogId(logId);
@ -96,12 +95,12 @@ public class CoordinatorLogId implements Serializable
static boolean isNone(int hostId, int hostLogId)
{
return hostId == NONE.hostId && hostLogId == NONE.hostLogId;
return hostId == NONE_HOST_ID && hostLogId == NONE_HOST_LOG_ID;
}
public boolean isNone()
{
return this == NONE || isNone(hostId, hostLogId);
return isNone(hostId, hostLogId);
}
@Override
@ -135,20 +134,12 @@ public class CoordinatorLogId implements Serializable
out.writeInt(logId.hostLogId);
}
public void serialize(long logId, DataOutputPlus out, int version) throws IOException
{
out.writeInt(hostId(logId));
out.writeInt(hostLogId(logId));
}
@Override
public CoordinatorLogId deserialize(DataInputPlus in, int version) throws IOException
{
int hostId = in.readInt();
int hostLogId = in.readInt();
if (isNone(hostId, hostLogId))
return none();
return new CoordinatorLogId(hostId, hostLogId);
return isNone(hostId, hostLogId) ? NONE : new CoordinatorLogId(hostId, hostLogId);
}
@Override
@ -156,11 +147,6 @@ public class CoordinatorLogId implements Serializable
{
return TypeSizes.sizeof(logId.hostId) + TypeSizes.sizeof(logId.hostLogId);
}
public long serializedSize(long logId, int version)
{
return TypeSizes.sizeof(logId);
}
}
public static final Serializer serializer = new Serializer();

8
src/java/org/apache/cassandra/replication/MutableCoordinatorLogOffsets.java
View File

@ -32,7 +32,7 @@ public interface MutableCoordinatorLogOffsets extends CoordinatorLogOffsets<Offs
{
for (long logId : mutations)
{
if (CoordinatorLogId.none().asLong() == logId)
if (CoordinatorLogId.NONE_LOG_ID == logId)
continue;
Offsets offsets = mutations.offsets(logId);
@ -42,10 +42,8 @@ public interface MutableCoordinatorLogOffsets extends CoordinatorLogOffsets<Offs
default void addAll(Offsets from)
{
if (from.logId().isNone())
return;
from.forEach(this::add);
if (!from.logId().isNone())
from.forEach(this::add);
}
static MutableCoordinatorLogOffsets create(boolean assumeExclusive)

35
src/java/org/apache/cassandra/replication/MutationId.java
View File

@ -26,6 +26,8 @@ import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import static org.apache.cassandra.replication.CoordinatorLogId.NONE_LOG_ID;
/**
* Full mutation id, with the addition of timestamp component.
* <p>
@ -34,11 +36,10 @@ import org.apache.cassandra.io.util.DataOutputPlus;
*/
public class MutationId extends ShortMutationId
{
private static final long NONE_LOG_ID = CoordinatorLogId.none().asLong();
private static final long NONE_SEQUENCE_ID = Long.MIN_VALUE;
private static final int NONE_OFFSET = offset(NONE_SEQUENCE_ID);
private static final int NONE_TIMESTAMP = timestamp(NONE_SEQUENCE_ID);
private static final MutationId NONE = new MutationId(NONE_LOG_ID, NONE_SEQUENCE_ID);
static final int NONE_TIMESTAMP = 0;
static final long NONE_SEQUENCE_ID = sequenceId(NONE_OFFSET, NONE_TIMESTAMP);
static final MutationId NONE = new MutationId(NONE_LOG_ID, NONE_SEQUENCE_ID);
/**
* 4 byte timestamp. The timestamp is monotonically non-decreasing.
@ -71,7 +72,7 @@ public class MutationId extends ShortMutationId
public static long sequenceId(int offset, int timestamp)
{
return ((long) offset << 32) | timestamp;
return ((long) offset << 32) | (timestamp & 0xffffffffL);
}
public static int offset(long sequenceId)
@ -95,17 +96,16 @@ public class MutationId extends ShortMutationId
return NONE;
}
@Override
public boolean isNone()
{
if (this == NONE)
return true;
return logId() == NONE_LOG_ID && offset() == NONE_OFFSET && timestamp() == NONE_TIMESTAMP;
return logId() == NONE_LOG_ID && sequenceId() == NONE_SEQUENCE_ID;
}
@Override
public String toString()
{
return "MutationId{" + hostId() + ", " + hostLogId() + ", " + offset() + ", " + timestamp() + '}';
return "MutationId{" + hostId + ", " + hostLogId + ", " + offset + ", " + timestamp + '}';
}
/**
@ -131,10 +131,9 @@ public class MutationId extends ShortMutationId
long logId = buffer.getLong(pos);
long sequenceId = buffer.getLong(pos + 8);
if (logId == MutationId.none().logId() && sequenceId == MutationId.none().sequenceId())
return MutationId.none();
return new MutationId(logId, sequenceId);
return logId == NONE_LOG_ID && sequenceId == NONE_SEQUENCE_ID
? NONE
: new MutationId(logId, sequenceId);
}
public static class Serializer implements IVersionedSerializer<MutationId>
@ -151,9 +150,9 @@ public class MutationId extends ShortMutationId
{
long logId = in.readLong();
long sequenceId = in.readLong();
if (logId == NONE_LOG_ID && sequenceId == NONE_SEQUENCE_ID)
return none();
return new MutationId(logId, sequenceId);
return logId == NONE_LOG_ID && sequenceId == NONE_SEQUENCE_ID
? NONE
: new MutationId(logId, sequenceId);
}
@Override
@ -167,7 +166,7 @@ public class MutationId extends ShortMutationId
in.readLong();
in.readLong();
}
};
}
public static Serializer serializer = new Serializer();
}

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

@ -338,7 +338,7 @@ public class MutationTrackingService
try
{
Preconditions.checkArgument(!mutationId.isNone());
Shard shard = getShardNullable(mutationId);
Shard shard = getShardNullable(mutationId.asLogId());
// A response to the coordinator (for a forwarded write) won't have the coordinator log matching it
if (shard != null)
shard.receivedWriteResponse(mutationId, fromHost);
@ -358,10 +358,7 @@ public class MutationTrackingService
logger.debug("{} receivedActivationAck from {}", transfer.logPrefix(), fromHost);
Preconditions.checkArgument(!transfer.id().isNone());
// REVIEW: This will be called with ShortMutationId, which overrides hashCode from CoordinatorLogId, but map
// is updated with CoordinatorLogId; shouldn't call this with a ShortMutationId, not sure why that's working
// elsewhere
Shard shard = getShardNullable(new CoordinatorLogId(transfer.id().logId()));
Shard shard = getShardNullable(transfer.id().asLogId());
// Local activation acknowledged in MutationTrackingService.activateLocal
if (shard != null && !fromHost.equals(FBUtilities.getBroadcastAddressAndPort()))
shard.receivedActivationResponse(transfer, fromHost);

2
src/java/org/apache/cassandra/replication/OutgoingMutations.java
View File

@ -107,7 +107,7 @@ class OutgoingMutations
private long key(int hostId, int offset)
{
return ((long) hostId << 32) | offset;
return ((long) hostId << 32) | (offset & 0xffffffffL);
}
private enum OutgoingStatus

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

@ -309,7 +309,7 @@ public class Shard
return getOrCreate(mutation.id());
}
private CoordinatorLog getOrCreate(MutationId mutationId)
private CoordinatorLog getOrCreate(ShortMutationId mutationId)
{
Preconditions.checkArgument(!mutationId.isNone());
return getOrCreate(mutationId.logId());

73
src/java/org/apache/cassandra/replication/ShortMutationId.java
View File

@ -18,6 +18,7 @@
package org.apache.cassandra.replication;
import java.io.IOException;
import java.io.Serializable;
import java.util.Comparator;
import org.apache.cassandra.db.TypeSizes;
@ -25,39 +26,67 @@ import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import static org.apache.cassandra.replication.CoordinatorLogId.NONE_HOST_ID;
import static org.apache.cassandra.replication.CoordinatorLogId.NONE_HOST_LOG_ID;
import static org.apache.cassandra.replication.CoordinatorLogId.NONE_LOG_ID;
/**
* MutationId without the timestamp component. This is sufficient for uniquely identifying a mutation,
* and for lookup in the journal and most tracking data structures.
*/
public class ShortMutationId extends CoordinatorLogId
public class ShortMutationId implements Serializable
{
static final int NONE_OFFSET = Integer.MIN_VALUE;
static final ShortMutationId NONE = new ShortMutationId(NONE_LOG_ID, NONE_OFFSET);
/** TCM host ID */
public final int hostId;
/**
* Host log ID (unique within the host).
* Allocated anew on host restart - one per token range replicated by the host.
* Persisted on allocation, unique within the host.
*/
public final int hostLogId;
/**
* 4 byte offset. Offest is incremented, is alone is sufficient to identify
* the entry within a coordinator log.
* MutationId adds a timestamp for correlation purposes.
*/
protected final int offset;
public final int offset;
public ShortMutationId(long logId, int offset)
{
super(logId);
this.offset = offset;
this(CoordinatorLogId.hostId(logId), CoordinatorLogId.hostLogId(logId), offset);
}
public ShortMutationId(CoordinatorLogId logId, int offset)
{
super(logId.hostId(), logId.hostLogId());
this(logId.hostId, logId.hostLogId, offset);
}
private ShortMutationId(int hostId, int hostLogId, int offset)
{
this.hostId = hostId;
this.hostLogId = hostLogId;
this.offset = offset;
}
public ShortMutationId(MutationId mutationId)
{
this(mutationId.logId(), mutationId.offset());
this(mutationId.hostLogId(), mutationId.hostId(), mutationId.offset());
}
public long logId()
public int hostId()
{
return super.asLong();
return hostId;
}
public int hostLogId()
{
return hostLogId;
}
public int offset()
@ -65,8 +94,23 @@ public class ShortMutationId extends CoordinatorLogId
return offset;
}
public long logId()
{
return CoordinatorLogId.asLong(hostId, hostLogId);
}
public CoordinatorLogId asLogId()
{
return new CoordinatorLogId(hostId, hostLogId);
}
public boolean isNone()
{
return hostId == NONE_HOST_ID && hostLogId == NONE_HOST_LOG_ID && offset == NONE_OFFSET;
}
@Override
public boolean equals(Object o)
public final boolean equals(Object o)
{
if (this == o) return true;
if (!(o instanceof ShortMutationId)) return false;
@ -75,9 +119,9 @@ public class ShortMutationId extends CoordinatorLogId
}
@Override
public int hashCode()
public final int hashCode()
{
return Integer.hashCode(offset) + 31 * super.hashCode();
return Integer.hashCode(offset) + 31 * Long.hashCode(logId());
}
@Override
@ -87,7 +131,7 @@ public class ShortMutationId extends CoordinatorLogId
}
public static final Comparator<ShortMutationId> comparator = (l, r) -> {
int cmp = CoordinatorLogId.comparator.compare(l, r);
int cmp = Long.compareUnsigned(l.logId(), r.logId());
return cmp != 0 ? cmp : Integer.compare(l.offset, r.offset);
};
@ -105,7 +149,10 @@ public class ShortMutationId extends CoordinatorLogId
{
long logId = in.readLong();
int offset = in.readInt();
return new ShortMutationId(logId, offset);
return (logId == NONE_LOG_ID && offset == NONE_OFFSET)
? NONE
: new ShortMutationId(logId, offset);
}
@Override

2
src/java/org/apache/cassandra/service/reads/tracked/TrackedLocalReads.java
View File

@ -179,7 +179,7 @@ public class TrackedLocalReads implements ExpiredStatePurger.Expireable
while (transferIds.hasNext())
{
ShortMutationId id = transferIds.next();
builder.builderForLog(id).unreconciled.add(id.offset());
builder.builderForLog(id.asLogId()).unreconciled.add(id.offset());
}
return builder.build();
}