Add deleted query param when looking up schema under subject (#168)

* Add deleted query param when looking up schema under subject

* Minor fix

* Minor fix

Author: rayokota

schemaregistry/mock-schemaregistry-client.ts

@@ -298,7 +298,8 @@ class MockClient implements Client {
     }
   }
 
-  async getVersion(subject: string, schema: SchemaInfo, normalize: boolean = false): Promise<number> {
+  async getVersion(subject: string, schema: SchemaInfo,
+                   normalize: boolean = false, deleted: boolean = false): Promise<number> {
     const cacheKey = stringify({ subject, schema: minimize(schema) });
     const cacheEntry = this.schemaToVersionCache.get(cacheKey);
 

schemaregistry/schemaregistry-client.ts

@@ -147,7 +147,7 @@ export interface Client {
   getLatestWithMetadata(subject: string, metadata: { [key: string]: string },
                         deleted: boolean, format?: string): Promise<SchemaMetadata>;
   getAllVersions(subject: string): Promise<number[]>;
-  getVersion(subject: string, schema: SchemaInfo, normalize: boolean): Promise<number>;
+  getVersion(subject: string, schema: SchemaInfo, normalize: boolean, deleted: boolean): Promise<number>;
   getAllSubjects(): Promise<string[]>;
   deleteSubject(subject: string, permanent: boolean): Promise<number[]>;
   deleteSubjectVersion(subject: string, version: number, permanent: boolean): Promise<number>;
@@ -436,8 +436,9 @@ export class SchemaRegistryClient implements Client {
    * @param schema - The schema for which to get the version.
    * @param normalize - Whether to normalize the schema before getting the version.
    */
-  async getVersion(subject: string, schema: SchemaInfo, normalize: boolean = false): Promise<number> {
-    const cacheKey = stringify({ subject, schema: minimize(schema) });
+  async getVersion(subject: string, schema: SchemaInfo,
+                   normalize: boolean = false, deleted: boolean = false): Promise<number> {
+    const cacheKey = stringify({ subject, schema: minimize(schema), deleted });
 
     return await this.schemaToVersionMutex.runExclusive(async () => {
       const cachedVersion: number | undefined = this.schemaToVersionCache.get(cacheKey);
@@ -448,7 +449,7 @@ export class SchemaRegistryClient implements Client {
       subject = encodeURIComponent(subject);
 
       const response: AxiosResponse<SchemaMetadata> = await this.restService.handleRequest(
-        `/subjects/${subject}?normalize=${normalize}`,
+        `/subjects/${subject}?normalize=${normalize}&deleted=${deleted}`,
         'POST',
         schema
       );

schemaregistry/serde/protobuf.ts

@@ -231,7 +231,7 @@ export class ProtobufSerializer extends Serializer implements ProtobufSerde {
         id = await this.client.getId(subject, info, normalize)
 
       }
-      version = await this.client.getVersion(subject, info, normalize)
+      version = await this.client.getVersion(subject, info, normalize, false)
     }
     return {
       id: id,

schemaregistry/serde/serde.ts

@@ -372,7 +372,7 @@ export abstract class Deserializer extends Serde {
 
   async getMigrations(subject: string, sourceInfo: SchemaInfo,
                 target: SchemaMetadata, format?: string): Promise<Migration[]> {
-    let version = await this.client.getVersion(subject, sourceInfo, false)
+    let version = await this.client.getVersion(subject, sourceInfo, false, true)
     let source: SchemaMetadata = {
       id: 0,
       version:    version,