diff --git a/src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java b/src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java
index 50cba9f90f6e..93a8252a983c 100644
--- a/src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java
+++ b/src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java
@@ -64,7 +64,8 @@ public void doVerb(Message<GossipDigestSyn> message)
             return;
         }
 
-        if (gDigestMessage.metadataId != ClusterMetadata.current().metadataIdentifier)
+        if (gDigestMessage.metadataId != ClusterMetadata.EMPTY_METADATA_IDENTIFIER
+            && gDigestMessage.metadataId != ClusterMetadata.current().metadataIdentifier)
         {
             logger.warn("Cluster metadata identifier mismatch from {} {}!={}", from, gDigestMessage.metadataId, ClusterMetadata.current().metadataIdentifier);
             return;
diff --git a/src/java/org/apache/cassandra/tcm/compatibility/GossipHelper.java b/src/java/org/apache/cassandra/tcm/compatibility/GossipHelper.java
index 8f4861cc17d2..31a20d0bd0d7 100644
--- a/src/java/org/apache/cassandra/tcm/compatibility/GossipHelper.java
+++ b/src/java/org/apache/cassandra/tcm/compatibility/GossipHelper.java
@@ -51,6 +51,8 @@
 import org.apache.cassandra.gms.VersionedValue;
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.schema.DistributedSchema;
+import org.apache.cassandra.schema.Keyspaces;
+import org.apache.cassandra.schema.SchemaConstants;
 import org.apache.cassandra.schema.SchemaKeyspace;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.tcm.ClusterMetadata;
@@ -275,9 +277,15 @@ private static NodeVersion getVersionFromEndpointState(InetAddressAndPort endpoi
 
     public static ClusterMetadata emptyWithSchemaFromSystemTables(Set<String> allKnownDatacenters)
     {
+        // If this instance was previously upgraded then subsequently downgraded, the metadata keyspace may have been
+        // added to system_schema tables. If so, don't include it in the initial schema as this will cause it to be
+        // incorrectly configured with the global partitioner. It will be created afresh from
+        // DistributedMetadataLogKeyspace.initialMetadata.
+        Keyspaces keyspaces = SchemaKeyspace.fetchNonSystemKeyspaces()
+                                            .filter(k -> !k.name.equals(SchemaConstants.METADATA_KEYSPACE_NAME));
         return new ClusterMetadata(Epoch.UPGRADE_STARTUP,
                                    DatabaseDescriptor.getPartitioner(),
-                                   DistributedSchema.fromSystemTables(SchemaKeyspace.fetchNonSystemKeyspaces(), allKnownDatacenters),
+                                   DistributedSchema.fromSystemTables(keyspaces, allKnownDatacenters),
                                    Directory.EMPTY,
                                    new TokenMap(DatabaseDescriptor.getPartitioner()),
                                    DataPlacements.empty(),