diff --git a/src/java/org/apache/cassandra/db/CounterMutation.java b/src/java/org/apache/cassandra/db/CounterMutation.java
index ed64e0aad7..14d7a010ff 100644
--- a/src/java/org/apache/cassandra/db/CounterMutation.java
+++ b/src/java/org/apache/cassandra/db/CounterMutation.java
@@ -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);
         }
diff --git a/src/java/org/apache/cassandra/db/Mutation.java b/src/java/org/apache/cassandra/db/Mutation.java
index cceb8ea510..5a41271474 100644
--- a/src/java/org/apache/cassandra/db/Mutation.java
+++ b/src/java/org/apache/cassandra/db/Mutation.java
@@ -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);
         }
diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java
index 84cc9bfa24..0f78bf8150 100644
--- a/src/java/org/apache/cassandra/db/ReadCommand.java
+++ b/src/java/org/apache/cassandra/db/ReadCommand.java
@@ -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)
diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogDescriptor.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogDescriptor.java
index 08bb189c90..82d8eb08df 100644
--- a/src/java/org/apache/cassandra/db/commitlog/CommitLogDescriptor.java
+++ b/src/java/org/apache/cassandra/db/commitlog/CommitLogDescriptor.java
@@ -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);
         }
diff --git a/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java b/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
index 9f25468495..5faf042bcb 100644
--- a/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
+++ b/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
@@ -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());
             }
         }
diff --git a/src/java/org/apache/cassandra/hints/HintsDescriptor.java b/src/java/org/apache/cassandra/hints/HintsDescriptor.java
index 02820bcd04..3079c0b80c 100644
--- a/src/java/org/apache/cassandra/hints/HintsDescriptor.java
+++ b/src/java/org/apache/cassandra/hints/HintsDescriptor.java
@@ -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();
         }
diff --git a/src/java/org/apache/cassandra/net/Message.java b/src/java/org/apache/cassandra/net/Message.java
index a20125c5a6..b0a0b9c48d 100644
--- a/src/java/org/apache/cassandra/net/Message.java
+++ b/src/java/org/apache/cassandra/net/Message.java
@@ -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);
diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java
index f4eaf72571..a0f40d3126 100644
--- a/src/java/org/apache/cassandra/net/MessagingService.java
+++ b/src/java/org/apache/cassandra/net/MessagingService.java
@@ -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);
diff --git a/src/java/org/apache/cassandra/tcm/membership/Directory.java b/src/java/org/apache/cassandra/tcm/membership/Directory.java
index ca5beecda5..bbd1225cbc 100644
--- a/src/java/org/apache/cassandra/tcm/membership/Directory.java
+++ b/src/java/org/apache/cassandra/tcm/membership/Directory.java
@@ -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;
             }
         }
diff --git a/test/data/serialization/5.1/gms.EndpointState.bin b/test/data/serialization/5.1/gms.EndpointState.bin
new file mode 100644
index 0000000000..d5c4eacd84
Binary files /dev/null and b/test/data/serialization/5.1/gms.EndpointState.bin differ
diff --git a/test/data/serialization/5.1/gms.Gossip.bin b/test/data/serialization/5.1/gms.Gossip.bin
new file mode 100644
index 0000000000..7a4fb5666e
Binary files /dev/null and b/test/data/serialization/5.1/gms.Gossip.bin differ
diff --git a/test/data/serialization/5.1/service.SyncComplete.bin b/test/data/serialization/5.1/service.SyncComplete.bin
new file mode 100644
index 0000000000..7c775cef66
Binary files /dev/null and b/test/data/serialization/5.1/service.SyncComplete.bin differ
diff --git a/test/data/serialization/5.1/service.SyncRequest.bin b/test/data/serialization/5.1/service.SyncRequest.bin
new file mode 100644
index 0000000000..f853b20f9c
Binary files /dev/null and b/test/data/serialization/5.1/service.SyncRequest.bin differ
diff --git a/test/data/serialization/5.1/service.ValidationComplete.bin b/test/data/serialization/5.1/service.ValidationComplete.bin
new file mode 100644
index 0000000000..f0c502458a
Binary files /dev/null and b/test/data/serialization/5.1/service.ValidationComplete.bin differ
diff --git a/test/data/serialization/5.1/service.ValidationRequest.bin b/test/data/serialization/5.1/service.ValidationRequest.bin
new file mode 100644
index 0000000000..04c492a8a1
Binary files /dev/null and b/test/data/serialization/5.1/service.ValidationRequest.bin differ
diff --git a/test/data/serialization/5.1/utils.EstimatedHistogram.bin b/test/data/serialization/5.1/utils.EstimatedHistogram.bin
new file mode 100644
index 0000000000..e878eda0ce
Binary files /dev/null and b/test/data/serialization/5.1/utils.EstimatedHistogram.bin differ
diff --git a/test/unit/org/apache/cassandra/AbstractSerializationsTester.java b/test/unit/org/apache/cassandra/AbstractSerializationsTester.java
index d1a96298bf..92d9cc9522 100644
--- a/test/unit/org/apache/cassandra/AbstractSerializationsTester.java
+++ b/test/unit/org/apache/cassandra/AbstractSerializationsTester.java
@@ -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();