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

Introduce 5.1 messaging format that brings in Epoch 

Patch by Alex Petrov; reviewed by Sam Tunnicliffe CASSANDRA-19066.
This commit is contained in:
Alex Petrov 2023-11-24 17:39:43 +01:00
parent a98a5a1d9d
commit e182744cd0
17 changed files with 54 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/java/org/apache/cassandra/db/CounterMutation.java
View File

@ -50,6 +50,7 @@ import org.apache.cassandra.utils.btree.BTreeSet;
import static java.util.concurrent.TimeUnit.*;
import static org.apache.cassandra.net.MessagingService.VERSION_40;
import static org.apache.cassandra.net.MessagingService.VERSION_50;
import static org.apache.cassandra.net.MessagingService.VERSION_51;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class CounterMutation implements IMutation
@ -332,6 +333,7 @@ public class CounterMutation implements IMutation
private int serializedSize40;
private int serializedSize50;
private int serializedSize51;
public int serializedSize(int version)
{
@ -345,6 +347,10 @@ public class CounterMutation implements IMutation
if (serializedSize50 == 0)
serializedSize50 = (int) serializer.serializedSize(this, VERSION_50);
return serializedSize50;
case VERSION_51:
if (serializedSize51 == 0)
serializedSize51 = (int) serializer.serializedSize(this, VERSION_51);
return serializedSize51;
default:
throw new IllegalStateException("Unknown serialization version: " + version);
}

7
src/java/org/apache/cassandra/db/Mutation.java
View File

@ -50,6 +50,7 @@ import org.apache.cassandra.utils.concurrent.Future;
import static org.apache.cassandra.net.MessagingService.VERSION_40;
import static org.apache.cassandra.net.MessagingService.VERSION_50;
import static org.apache.cassandra.net.MessagingService.VERSION_51;
import static org.apache.cassandra.utils.MonotonicClock.Global.approxTime;
public class Mutation implements IMutation, Supplier<Mutation>
@ -317,6 +318,7 @@ public class Mutation implements IMutation, Supplier<Mutation>
private int serializedSize40;
private int serializedSize50;
private int serializedSize51;
public int serializedSize(int version)
{
@ -330,7 +332,10 @@ public class Mutation implements IMutation, Supplier<Mutation>
if (serializedSize50 == 0)
serializedSize50 = (int) serializer.serializedSize(this, VERSION_50);
return serializedSize50;
case VERSION_51:
if (serializedSize51 == 0)
serializedSize51 = (int) serializer.serializedSize(this, VERSION_51);
return serializedSize51;
default:
throw new IllegalStateException("Unknown serialization version: " + version);
}

4
src/java/org/apache/cassandra/db/ReadCommand.java
View File

@ -1099,7 +1099,7 @@ public abstract class ReadCommand extends AbstractReadQuery
if (command.isDigestQuery())
out.writeUnsignedVInt32(command.digestVersion());
command.metadata().id.serialize(out);
if (version >= MessagingService.VERSION_50)
if (version >= MessagingService.VERSION_51)
Epoch.serializer.serialize(command.serializedAtEpoch, out);
out.writeInt(version >= MessagingService.VERSION_50 ? CassandraUInt.fromLong(command.nowInSec()) : (int) command.nowInSec());
ColumnFilter.serializer.serialize(command.columnFilter(), out, version);
@ -1189,7 +1189,7 @@ public abstract class ReadCommand extends AbstractReadQuery
return 2 // kind + flags
+ (command.isDigestQuery() ? TypeSizes.sizeofUnsignedVInt(command.digestVersion()) : 0)
+ command.metadata().id.serializedSize()
+ (version >= MessagingService.VERSION_50 ? Epoch.serializer.serializedSize(command.metadata().epoch) : 0)
+ (version >= MessagingService.VERSION_51 ? Epoch.serializer.serializedSize(command.metadata().epoch) : 0)
+ TypeSizes.INT_SIZE // command.nowInSec() is serialized as uint
+ ColumnFilter.serializer.serializedSize(command.columnFilter(), version)
+ RowFilter.serializer.serializedSize(command.rowFilter(), version)

5
src/java/org/apache/cassandra/db/commitlog/CommitLogDescriptor.java
View File

@ -63,13 +63,14 @@ public class CommitLogDescriptor
public static final int VERSION_30 = 6;
public static final int VERSION_40 = 7;
public static final int VERSION_50 = 8;
public static final int VERSION_51 = 9;
/**
* Increment this number if there is a changes in the commit log disc layout or MessagingVersion changes.
* Note: make sure to handle {@link #getMessagingVersion()}
*/
@VisibleForTesting
public static final int current_version = DatabaseDescriptor.getStorageCompatibilityMode().isBefore(5) ? VERSION_40 : VERSION_50;
public static final int current_version = DatabaseDescriptor.getStorageCompatibilityMode().isBefore(5) ? VERSION_40 : VERSION_51;
final int version;
public final long id;
@ -225,6 +226,8 @@ public class CommitLogDescriptor
return MessagingService.VERSION_40;
case VERSION_50:
return MessagingService.VERSION_50;
case VERSION_51:
return MessagingService.VERSION_51;
default:
throw new IllegalStateException("Unknown commitlog version " + version);
}

8
src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
View File

@ -729,7 +729,7 @@ public class PartitionUpdate extends AbstractBTreePartition
assert !iter.isReverseOrder();
update.metadata.id.serialize(out);
if (version >= MessagingService.VERSION_50)
if (version >= MessagingService.VERSION_51)
Epoch.serializer.serialize(update.metadata.epoch != null ? update.metadata.epoch : Epoch.EMPTY, out);
UnfilteredRowIteratorSerializer.serializer.serialize(iter, null, out, version, update.rowCount());
}
@ -739,7 +739,7 @@ public class PartitionUpdate extends AbstractBTreePartition
{
TableId tableId = TableId.deserialize(in);
Epoch remoteVersion = null;
if (version >= MessagingService.VERSION_50)
if (version >= MessagingService.VERSION_51)
remoteVersion = Epoch.serializer.deserialize(in);
TableMetadata tableMetadata;
try
@ -796,7 +796,7 @@ public class PartitionUpdate extends AbstractBTreePartition
if (position >= in.limit())
throw new EOFException();
if (version >= MessagingService.VERSION_50)
if (version >= MessagingService.VERSION_51)
{
long epoch = VIntCoding.getUnsignedVInt(in, position);
position += VIntCoding.computeVIntSize(epoch);
@ -816,7 +816,7 @@ public class PartitionUpdate extends AbstractBTreePartition
try (UnfilteredRowIterator iter = update.unfilteredIterator())
{
return update.metadata.id.serializedSize()
+ (version >= MessagingService.VERSION_50 ? Epoch.serializer.serializedSize(update.metadata.epoch) : 0)
+ (version >= MessagingService.VERSION_51 ? Epoch.serializer.serializedSize(update.metadata.epoch) : 0)
+ UnfilteredRowIteratorSerializer.serializer.serializedSize(iter, null, version, update.rowCount());
}
}

5
src/java/org/apache/cassandra/hints/HintsDescriptor.java
View File

@ -69,7 +69,8 @@ final class HintsDescriptor
static final int VERSION_30 = 1;
static final int VERSION_40 = 2;
static final int VERSION_50 = 3;
static final int CURRENT_VERSION = DatabaseDescriptor.getStorageCompatibilityMode().isBefore(5) ? VERSION_40 : VERSION_50;
static final int VERSION_51 = 4;
static final int CURRENT_VERSION = DatabaseDescriptor.getStorageCompatibilityMode().isBefore(5) ? VERSION_40 : VERSION_51;
static final String COMPRESSION = "compression";
static final String ENCRYPTION = "encryption";
@ -237,6 +238,8 @@ final class HintsDescriptor
return MessagingService.VERSION_40;
case VERSION_50:
return MessagingService.VERSION_50;
case VERSION_51:
return MessagingService.VERSION_51;
default:
throw new AssertionError();
}

27
src/java/org/apache/cassandra/net/Message.java
View File

@ -53,6 +53,7 @@ import static org.apache.cassandra.db.TypeSizes.sizeof;
import static org.apache.cassandra.db.TypeSizes.sizeofUnsignedVInt;
import static org.apache.cassandra.net.MessagingService.VERSION_40;
import static org.apache.cassandra.net.MessagingService.VERSION_50;
import static org.apache.cassandra.net.MessagingService.VERSION_51;
import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
import static org.apache.cassandra.utils.MonotonicClock.Global.approxTime;
import static org.apache.cassandra.utils.vint.VIntCoding.*;
@ -826,7 +827,7 @@ public class Message<T>
return -1; // not enough bytes to read id
index += idSize;
if (version >= VERSION_50)
if (version >= VERSION_51)
{
int epochSize = computeUnsignedVIntSize(buf, index, readerLimit);
if (epochSize < 0)
@ -888,7 +889,7 @@ public class Message<T>
index += computeUnsignedVIntSize(id);
Epoch epoch = Epoch.EMPTY;
if (version >= VERSION_50)
if (version >= VERSION_51)
{
long epochl = getUnsignedVInt(buf, index);
index += computeUnsignedVIntSize(epochl);
@ -924,7 +925,7 @@ public class Message<T>
private void serializeHeader(Header header, DataOutputPlus out, int version) throws IOException
{
out.writeUnsignedVInt(header.id);
if (version >= VERSION_50)
if (version >= VERSION_51)
Epoch.messageSerializer.serialize(header.epoch, out, version);
// int cast cuts off the high-order half of the timestamp, which we can assume remains
// the same between now and when the recipient reconstructs it.
@ -939,7 +940,7 @@ public class Message<T>
{
long id = in.readUnsignedVInt();
Epoch epoch = Epoch.EMPTY;
if (version >= VERSION_50)
if (version >= VERSION_51)
epoch = Epoch.messageSerializer.deserialize(in, version);
long currentTimeNanos = approxTime.now();
MonotonicClockTranslation timeSnapshot = approxTime.translate();
@ -955,7 +956,7 @@ public class Message<T>
private void skipHeader(DataInputPlus in, int version) throws IOException
{
skipUnsignedVInt(in); // id
if (version >= VERSION_50)
if (version >= VERSION_51)
skipUnsignedVInt(in); // epoch
in.skipBytesFully(4); // createdAt
skipUnsignedVInt(in); // expiresIn
@ -968,7 +969,7 @@ public class Message<T>
{
long size = 0;
size += sizeofUnsignedVInt(header.id);
if (version >= VERSION_50)
if (version >= VERSION_51)
size += sizeofUnsignedVInt(header.epoch.getEpoch());
size += CREATION_TIME_SIZE;
size += sizeofUnsignedVInt(NANOSECONDS.toMillis(header.expiresAtNanos - header.createdAtNanos));
@ -1164,6 +1165,7 @@ public class Message<T>
private int serializedSize40;
private int serializedSize50;
private int serializedSize51;
/**
* Serialized size of the entire message, for the provided messaging version. Caches the calculated value.
@ -1180,13 +1182,18 @@ public class Message<T>
if (serializedSize50 == 0)
serializedSize50 = serializer.serializedSize(this, VERSION_50);
return serializedSize50;
case VERSION_51:
if (serializedSize51 == 0)
serializedSize51 = serializer.serializedSize(this, VERSION_51);
return serializedSize51;
default:
throw new IllegalStateException("Unknown serialization version " + version);
}
}
private int payloadSize40 = -1;
private int payloadSize50 = -1;
private int payloadSize40 = -1;
private int payloadSize50 = -1;
private int payloadSize51 = -1;
private int payloadSize(int version)
{
@ -1200,6 +1207,10 @@ public class Message<T>
if (payloadSize50 < 0)
payloadSize50 = serializer.payloadSize(this, VERSION_50);
return payloadSize50;
case VERSION_51:
if (payloadSize51 < 0)
payloadSize51 = serializer.payloadSize(this, VERSION_51);
return payloadSize51;
default:
throw new IllegalStateException("Unkown serialization version " + version);

6
src/java/org/apache/cassandra/net/MessagingService.java
View File

@ -221,9 +221,10 @@ public class MessagingService extends MessagingServiceMBeanImpl implements Messa
VERSION_3014(11),
VERSION_40(12),
// c14227 TTL overflow, 'uint' timestamps
VERSION_50(13);
VERSION_50(13),
VERSION_51(14);
public static final Version CURRENT = DatabaseDescriptor.getStorageCompatibilityMode().isBefore(5) ? VERSION_40 : VERSION_50;
public static final Version CURRENT = DatabaseDescriptor.getStorageCompatibilityMode().isBefore(5) ? VERSION_40 : VERSION_51;
public final int value;
@ -253,6 +254,7 @@ public class MessagingService extends MessagingServiceMBeanImpl implements Messa
public static final int VERSION_3014 = 11;
public static final int VERSION_40 = 12;
public static final int VERSION_50 = 13; // c14227 TTL overflow, 'uint' timestamps
public static final int VERSION_51 = 14; // TCM
public static final int minimum_version = VERSION_40;
public static final int current_version = Version.CURRENT.value;
static AcceptVersions accept_messaging = new AcceptVersions(minimum_version, current_version);

6
src/java/org/apache/cassandra/tcm/membership/Directory.java
View File

@ -487,7 +487,7 @@ public class Directory implements MetadataValue<Directory>
else
{
out.writeBoolean(true);
UUIDSerializer.serializer.serialize(node.hostId, out, MessagingService.VERSION_50);
UUIDSerializer.serializer.serialize(node.hostId, out, MessagingService.VERSION_51);
}
}
@ -500,7 +500,7 @@ public class Directory implements MetadataValue<Directory>
NodeState state = NodeState.values()[in.readInt()];
NodeVersion nodeVersion = NodeVersion.serializer.deserialize(in, version);
boolean hasHostId = in.readBoolean();
UUID hostId = hasHostId ? UUIDSerializer.serializer.deserialize(in, MessagingService.VERSION_50) : null;
UUID hostId = hasHostId ? UUIDSerializer.serializer.deserialize(in, MessagingService.VERSION_51) : null;
return new Node(id, addresses, location, state, nodeVersion, hostId);
}
@ -515,7 +515,7 @@ public class Directory implements MetadataValue<Directory>
size += NodeVersion.serializer.serializedSize(node.version, version);
size += TypeSizes.BOOL_SIZE;
if (node.hostId != null)
size += UUIDSerializer.serializer.serializedSize(node.hostId, MessagingService.VERSION_50);
size += UUIDSerializer.serializer.serializedSize(node.hostId, MessagingService.VERSION_51);
return size;
}
}

BIN
test/data/serialization/5.1/gms.EndpointState.bin Normal file
View File

Binary file not shown.

BIN
test/data/serialization/5.1/gms.Gossip.bin Normal file
View File

Binary file not shown.

BIN
test/data/serialization/5.1/service.SyncComplete.bin Normal file
View File

Binary file not shown.

BIN
test/data/serialization/5.1/service.SyncRequest.bin Normal file
View File

Binary file not shown.

BIN
test/data/serialization/5.1/service.ValidationComplete.bin Normal file
View File

Binary file not shown.

BIN
test/data/serialization/5.1/service.ValidationRequest.bin Normal file
View File

Binary file not shown.

BIN
test/data/serialization/5.1/utils.EstimatedHistogram.bin Normal file
View File

Binary file not shown.

3
test/unit/org/apache/cassandra/AbstractSerializationsTester.java
View File

@ -32,12 +32,13 @@ import static org.apache.cassandra.config.CassandraRelevantProperties.TEST_SERIA
public class AbstractSerializationsTester
{
protected static final String CUR_VER = CASSANDRA_VERSION.getString("5.0");
protected static final String CUR_VER = CASSANDRA_VERSION.getString("5.1");
protected static final Map<String, Integer> VERSION_MAP = new HashMap<String, Integer> ()
{{
put("3.0", MessagingService.VERSION_30);
put("4.0", MessagingService.VERSION_40);
put("5.0", MessagingService.VERSION_50);
put("5.1", MessagingService.VERSION_51);
}};
protected static final boolean EXECUTE_WRITES = TEST_SERIALIZATION_WRITES.getBoolean();