diff --git a/CHANGES.txt b/CHANGES.txt
index 23dbee1941..801c6f2e48 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 4.1.5
+ * Concurrent equivalent schema updates lead to unresolved disagreement (CASSANDRA-19578)
  * Fix hints delivery for a node going down repeatedly (CASSANDRA-19495)
  * Do not go to disk for reading hints file sizes (CASSANDRA-19477)
  * Fix system_views.settings to handle array types (CASSANDRA-19475)
diff --git a/src/java/org/apache/cassandra/schema/DefaultSchemaUpdateHandler.java b/src/java/org/apache/cassandra/schema/DefaultSchemaUpdateHandler.java
index 0f0c3e9cf5..1a654d0ff9 100644
--- a/src/java/org/apache/cassandra/schema/DefaultSchemaUpdateHandler.java
+++ b/src/java/org/apache/cassandra/schema/DefaultSchemaUpdateHandler.java
@@ -197,7 +197,8 @@ public class DefaultSchemaUpdateHandler implements SchemaUpdateHandler, IEndpoin
         // no-op
     }
 
-    private synchronized SchemaTransformationResult applyMutations(Collection<Mutation> schemaMutations)
+    @VisibleForTesting
+    synchronized SchemaTransformationResult applyMutations(Collection<Mutation> schemaMutations)
     {
         // fetch the current state of schema for the affected keyspaces only
         DistributedSchema before = schema;
@@ -253,7 +254,7 @@ public class DefaultSchemaUpdateHandler implements SchemaUpdateHandler, IEndpoin
 
     private void updateSchema(SchemaTransformationResult update, boolean local)
     {
-        if (!update.diff.isEmpty())
+        if (!update.diff.isEmpty() || !update.after.getVersion().equals(schema.getVersion()))
         {
             this.schema = update.after;
             logger.debug("Schema updated: {}", update);
diff --git a/test/unit/org/apache/cassandra/schema/SchemaTest.java b/test/unit/org/apache/cassandra/schema/SchemaTest.java
index 81f409f477..1c6b6aa483 100644
--- a/test/unit/org/apache/cassandra/schema/SchemaTest.java
+++ b/test/unit/org/apache/cassandra/schema/SchemaTest.java
@@ -89,6 +89,24 @@ public class SchemaTest
         }
     }
 
+    @Test
+    public void testSchemaVersionUpdates() {
+        KeyspaceMetadata ksm = KeyspaceMetadata.create("testSchemaVersionUpdates", KeyspaceParams.simple(1));
+        SchemaTransformation.SchemaTransformationResult r = Schema.instance.transform(current -> current.withAddedOrUpdated(ksm));
+
+        Collection<Mutation> mutations = SchemaKeyspace.convertSchemaDiffToMutations(r.diff, FBUtilities.timestampMicros());
+        if (Schema.instance.updateHandler instanceof DefaultSchemaUpdateHandler)
+        {
+            ((DefaultSchemaUpdateHandler) Schema.instance.updateHandler).applyMutations(mutations);
+            // now as if a 2nd schema update is received with same data (ie same alter to 2 differnet nodes)
+            mutations = SchemaKeyspace.convertSchemaDiffToMutations(r.diff, FBUtilities.timestampMicros());
+            ((DefaultSchemaUpdateHandler) Schema.instance.updateHandler).applyMutations(mutations);
+            // schema should match the current digest
+            assertEquals(SchemaKeyspace.calculateSchemaDigest(), Schema.instance.getVersion());
+        }
+        Schema.instance.transform(current -> current.without(ksm.name));
+    }
+
     @Test
     public void testKeyspaceCreationWhenNotInitialized() {
         Keyspace.unsetInitialized();