diff --git a/CHANGES.txt b/CHANGES.txt
index 060fa9d56b..9ba36fafb5 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,6 +1,7 @@
 3.0.18
  * RangeTombstoneList doesn't properly clean up mergeable or superseded rts in some cases (CASSANDRA-14894)
  * Fix handling of collection tombstones for dropped columns from legacy sstables (CASSANDRA-14912)
+ * Throw exception if Columns serialized subset encode more columns than possible (CASSANDRA-14591)
  * Drop/add column name with different Kind can result in corruption (CASSANDRA-14843)
  * Fix missing rows when reading 2.1 SSTables with static columns in 3.0 (CASSANDRA-14873)
  * Move TWCS message 'No compaction necessary for bucket size' to Trace level (CASSANDRA-14884)
diff --git a/src/java/org/apache/cassandra/db/Columns.java b/src/java/org/apache/cassandra/db/Columns.java
index 45ce91ec0a..18e17d7eda 100644
--- a/src/java/org/apache/cassandra/db/Columns.java
+++ b/src/java/org/apache/cassandra/db/Columns.java
@@ -526,6 +526,8 @@ public class Columns extends AbstractCollection<ColumnDefinition> implements Col
                     }
                     encoded >>>= 1;
                 }
+                if (encoded != 0)
+                    throw new IOException("Invalid Columns subset bytes; too many bits set:" + Long.toBinaryString(encoded));
                 return new Columns(builder.build(), firstComplexIdx);
             }
         }
diff --git a/test/unit/org/apache/cassandra/db/ColumnsTest.java b/test/unit/org/apache/cassandra/db/ColumnsTest.java
index 1a245a025d..9498e8b756 100644
--- a/test/unit/org/apache/cassandra/db/ColumnsTest.java
+++ b/test/unit/org/apache/cassandra/db/ColumnsTest.java
@@ -27,6 +27,8 @@ import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterators;
 import com.google.common.collect.Lists;
 
+import org.apache.cassandra.cql3.ColumnIdentifier;
+import org.apache.cassandra.db.marshal.BytesType;
 import org.junit.AfterClass;
 import org.junit.Test;
 
@@ -48,6 +50,31 @@ public class ColumnsTest
 
     private static final CFMetaData cfMetaData = MockSchema.newCFS().metadata;
 
+    @Test
+    public void testDeserializeCorruption() throws IOException
+    {
+        ColumnsCheck check = randomSmall(1, 0, 3, 0);
+        Columns superset = check.columns;
+        List<ColumnDefinition> minus1 = new ArrayList<>(check.definitions);
+        minus1.remove(3);
+        Columns minus2 = check.columns
+                .without(check.columns.getSimple(3))
+                .without(check.columns.getSimple(2));
+        try (DataOutputBuffer out = new DataOutputBuffer())
+        {
+            // serialize a subset
+            Columns.serializer.serializeSubset(minus1, superset, out);
+            try (DataInputBuffer in = new DataInputBuffer(out.toByteArray()))
+            {
+                Columns.serializer.deserializeSubset(minus2, in);
+                Assert.assertFalse(true);
+            }
+            catch (IOException e)
+            {
+            }
+        }
+    }
+
     // this tests most of our functionality, since each subset we perform
     // reasonably comprehensive tests of basic functionality against
     @Test
@@ -455,7 +482,7 @@ public class ColumnsTest
             results.add(ColumnDefinition.regularDef(cfMetaData, bytes(name), UTF8Type.instance));
     }
 
-    private static <V> void addComplex(List<String> names, List<ColumnDefinition> results)
+    private static void addComplex(List<String> names, List<ColumnDefinition> results)
     {
         for (String name : names)
             results.add(ColumnDefinition.regularDef(cfMetaData, bytes(name), SetType.getInstance(UTF8Type.instance, true)));