Merge branch '8.1.x' into master by rayokota

Author: semaphore-agent-production[bot]

client/src/main/java/io/confluent/kafka/schemaregistry/client/MockSchemaRegistryClient.java

@@ -519,7 +519,7 @@ public int getVersion(String subject, ParsedSchema schema, boolean normalize)
       schema = schema.normalize();
     }
     Map<ParsedSchema, Integer> versions = schemaToVersionCache.get(subject);
-    if (versions != null) {
+    if (versions != null && versions.containsKey(schema)) {
       return versions.get(schema);
     } else {
       throw new RestClientException("Subject Not Found", 404, 40401);

core/src/test/java/io/confluent/kafka/schemaregistry/client/MockSchemaRegistryClientTest.java

@@ -19,6 +19,7 @@
 import io.confluent.kafka.schemaregistry.avro.AvroSchema;
 import io.confluent.kafka.schemaregistry.avro.AvroSchemaProvider;
 import io.confluent.kafka.schemaregistry.client.rest.entities.SubjectVersion;
+import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;
 import java.time.Duration;
 import java.util.Collection;
 import java.util.Collections;
@@ -41,6 +42,7 @@
 import java.util.Properties;
 import org.junit.jupiter.api.Test;
 
+import static org.junit.Assert.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
@@ -183,6 +185,8 @@ public void testRegisterAndGetId() throws Exception {
     int id = client.register("test", schema);
     assertEquals(1, id);
 
+    assertThrows(RestClientException.class, () -> client.getVersion("test", schema2));
+
     id = client.register("test", schema2);
     assertEquals(2, id);