diff --git a/client/src/main/java/io/confluent/kafka/schemaregistry/client/MockSchemaRegistryClient.java b/client/src/main/java/io/confluent/kafka/schemaregistry/client/MockSchemaRegistryClient.java
index 8d52d3dcab2..1f0012f0214 100644
--- a/client/src/main/java/io/confluent/kafka/schemaregistry/client/MockSchemaRegistryClient.java
+++ b/client/src/main/java/io/confluent/kafka/schemaregistry/client/MockSchemaRegistryClient.java
@@ -348,11 +348,18 @@ public Collection<String> getAllSubjectsById(int id) {
   public Collection<SubjectVersion> getAllVersionsById(int id) {
     return idToSchemaCache.entrySet().stream()
         .filter(entry -> entry.getValue().containsKey(id))
-        .map(e -> {
+        .flatMap(e -> {
           ParsedSchema schema = e.getValue().get(id);
-          int version = schemaToVersionCache.get(e.getKey()).get(schema);
-          return new SubjectVersion(e.getKey(), version);
-        }).collect(Collectors.toList());
+          Map<ParsedSchema, Integer> schemaVersionMap = schemaToVersionCache.get(e.getKey());
+          if (schemaVersionMap != null) {
+            int version = schemaVersionMap.get(schema);
+            return Stream.of(new SubjectVersion(e.getKey(), version));
+          } else {
+            return Stream.empty();
+          }
+        })
+        .distinct()
+        .collect(Collectors.toList());
   }
 
   private int getLatestVersion(String subject)
diff --git a/core/src/test/java/io/confluent/kafka/schemaregistry/client/MockSchemaRegistryClientTest.java b/core/src/test/java/io/confluent/kafka/schemaregistry/client/MockSchemaRegistryClientTest.java
index 80c7659b42b..761fc2520ea 100644
--- a/core/src/test/java/io/confluent/kafka/schemaregistry/client/MockSchemaRegistryClientTest.java
+++ b/core/src/test/java/io/confluent/kafka/schemaregistry/client/MockSchemaRegistryClientTest.java
@@ -18,6 +18,8 @@
 import io.confluent.kafka.schemaregistry.ClusterTestHarness;
 import io.confluent.kafka.schemaregistry.avro.AvroSchema;
 import io.confluent.kafka.schemaregistry.avro.AvroSchemaProvider;
+import io.confluent.kafka.schemaregistry.client.rest.entities.SubjectVersion;
+import java.util.Collection;
 import java.util.Collections;
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericData;
@@ -200,5 +202,16 @@ public void testRegisterAndGetId() throws Exception {
     id = client.getId("test2", schema2);
     assertEquals(2, id);
   }
+
+  @Test
+  public void testGetAllVersionsById() throws Exception {
+    AvroSchema avroSchema = new AvroSchema("{\"type\":\"record\",\"name\":\"ts1\","
+        + "\"fields\":[{\"name\": \"fld1\",\"type\": \"int\"}]}");
+    MockSchemaRegistryClient client =
+        new MockSchemaRegistryClient(Collections.singletonList(new AvroSchemaProvider()));
+    int id = client.register("test-value", avroSchema);
+    Collection<SubjectVersion> versions = client.getAllVersionsById(id);
+    assertEquals(new SubjectVersion("test-value", 1), versions.iterator().next());
+  }
 }