diff --git a/CHANGES.txt b/CHANGES.txt
index b0aad8c185..1d6a84c787 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -5,6 +5,7 @@
 
 
 3.0
+ * Fix bug when adding a column to thrift with the same name than a primary key (CASSANDRA-10608)
  * Add client address argument to IAuthenticator::newSaslNegotiator (CASSANDRA-8068)
  * Fix implementation of LegacyLayout.LegacyBoundComparator (CASSANDRA-10602)
  * Don't use 'names query' read path for counters (CASSANDRA-10572)
diff --git a/src/java/org/apache/cassandra/db/LegacyLayout.java b/src/java/org/apache/cassandra/db/LegacyLayout.java
index 3c64443e70..91b7755f5c 100644
--- a/src/java/org/apache/cassandra/db/LegacyLayout.java
+++ b/src/java/org/apache/cassandra/db/LegacyLayout.java
@@ -157,13 +157,16 @@ public abstract class LegacyLayout
             return new LegacyCellName(clustering, null, null);
 
         ColumnDefinition def = metadata.getColumnDefinition(column);
-        if (def == null)
+        if ((def == null) || def.isPrimaryKeyColumn())
         {
             // If it's a compact table, it means the column is in fact a "dynamic" one
             if (metadata.isCompactTable())
                 return new LegacyCellName(new Clustering(column), metadata.compactValueColumn(), null);
 
-            throw new UnknownColumnException(metadata, column);
+            if (def == null)
+                throw new UnknownColumnException(metadata, column);
+            else
+                throw new IllegalArgumentException("Cannot add primary key column to partition update");
         }
 
         ByteBuffer collectionElement = metadata.isCompound() ? CompositeType.extractComponent(cellname, metadata.comparator.size() + 1) : null;
diff --git a/test/unit/org/apache/cassandra/db/LegacyCellNameTest.java b/test/unit/org/apache/cassandra/db/LegacyCellNameTest.java
new file mode 100644
index 0000000000..fa29b1e3fb
--- /dev/null
+++ b/test/unit/org/apache/cassandra/db/LegacyCellNameTest.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.db;
+
+import org.junit.Test;
+
+import org.apache.cassandra.config.CFMetaData;
+
+import static junit.framework.Assert.assertTrue;
+
+public class LegacyCellNameTest
+{
+    @Test
+    public void testColumnSameNameAsPartitionKeyCompactStorage() throws Exception
+    {
+        CFMetaData cfm = CFMetaData.compile("CREATE TABLE cs (" +
+                                            "k int PRIMARY KEY, v int)" +
+                                            " WITH COMPACT STORAGE", "ks");
+
+        LegacyLayout.LegacyCellName cellName 
+            = LegacyLayout.decodeCellName(cfm, 
+                                          LegacyLayout.makeLegacyComparator(cfm)
+                                                      .fromString("k"));
+
+        assertTrue(cellName.column.isRegular());
+    }
+
+    @Test
+    public void testColumnSameNameAsClusteringKeyCompactStorage() throws Exception
+    {
+        CFMetaData cfm = CFMetaData.compile("CREATE TABLE cs (" +
+                                            "k int PRIMARY KEY, v int)" +
+                                            " WITH COMPACT STORAGE", "ks");
+
+        LegacyLayout.LegacyCellName cellName 
+            = LegacyLayout.decodeCellName(cfm, 
+                                          LegacyLayout.makeLegacyComparator(cfm)
+                                                      .fromString("column1"));
+
+        assertTrue(cellName.column.isRegular());
+    }
+
+    @Test(expected=IllegalArgumentException.class)
+    public void testColumnSameNameAsPartitionKeyCql3() throws Exception
+    {
+        CFMetaData cfm = CFMetaData.compile("CREATE TABLE cs (" +
+                                            "k int PRIMARY KEY, v int)", "ks");
+
+        LegacyLayout.LegacyCellName cellName 
+            = LegacyLayout.decodeCellName(cfm, 
+                                          LegacyLayout.makeLegacyComparator(cfm)
+                                                      .fromString("k"));
+    }
+
+    @Test(expected=IllegalArgumentException.class)
+    public void testColumnSameNameAsClusteringKeyCql3() throws Exception
+    {
+        CFMetaData cfm = CFMetaData.compile("CREATE TABLE cs (" +
+                                            "k int, c text, v int, PRIMARY KEY(k, c))", "ks");
+
+        LegacyLayout.LegacyCellName cellName 
+            = LegacyLayout.decodeCellName(cfm, 
+                                          LegacyLayout.makeLegacyComparator(cfm)
+                                                      .fromString("c"));
+    }
+}