Prevent Unnecessary Requests on Cache Misses (#48)

* Limit cache miss requests to single in-flight * Use Node 12.x in Azure Pipelines * Update src/SchemaRegistry.ts Co-Authored-By: Erik Engervall <[email protected]> * Update src/SchemaRegistry.ts Co-Authored-By: Erik Engervall <[email protected]> * Update src/SchemaRegistry.ts Co-Authored-By: Erik Engervall <[email protected]>
kafkajs · Mar 18, 2020 · c03ecb7 · c03ecb7
1 parent 67e45e8
commit c03ecb7
Show file tree

Hide file tree

Showing 3 changed files with 43 additions and 6 deletions.
diff --git a/azure-pipelines.yml b/azure-pipelines.yml
@@ -25,7 +25,7 @@ jobs:
     steps:
       - task: NodeTool@0
         inputs:
-          versionSpec: '8.x'
+          versionSpec: '12.x'
       - bash: yarn install
         displayName: yarn install
       - bash: yarn lint
@@ -38,7 +38,7 @@ jobs:
     steps:
       - task: NodeTool@0
         inputs:
-          versionSpec: '8.x'
+          versionSpec: '12.x'
       - bash: yarn install
         displayName: yarn install
       - bash: yarn build
@@ -52,7 +52,7 @@ jobs:
     steps:
       - task: NodeTool@0
         inputs:
-          versionSpec: '8.x'
+          versionSpec: '12.x'
       - bash: yarn install
         displayName: yarn install
       - bash: docker-compose -f ${COMPOSE_FILE} pull
@@ -72,7 +72,7 @@ jobs:
     steps:
       - task: NodeTool@0
         inputs:
-          versionSpec: '8.x'
+          versionSpec: '12.x'
       - bash: yarn install
         displayName: yarn install
       - bash: yarn prepare:release
@@ -99,7 +99,7 @@ jobs:
     steps:
       - task: NodeTool@0
         inputs:
-          versionSpec: '8.x'
+          versionSpec: '12.x'
       - bash: git config core.autocrlf true || test true
         displayName: git config core.autocrlf
       - bash: git config --global user.name "${GH_NAME}" || test true

diff --git a/src/SchemaRegistry.spec.ts b/src/SchemaRegistry.spec.ts
@@ -167,6 +167,22 @@ describe('SchemaRegistry', () => {
       expect(schemaRegistry.cache.getSchema(registryId)).toBeTruthy()
     })
 
+    it('creates a single origin request for a schema cache-miss', async () => {
+      const buffer = Buffer.from(await schemaRegistry.encode(registryId, payload))
+
+      schemaRegistry.cache.clear()
+
+      const spy = jest.spyOn((schemaRegistry as any).api.Schema, 'find')
+
+      await Promise.all([
+        schemaRegistry.decode(buffer),
+        schemaRegistry.decode(buffer),
+        schemaRegistry.decode(buffer),
+      ])
+
+      expect(spy).toHaveBeenCalledTimes(1)
+    })
+
     describe('when the cache is populated', () => {
       it('uses the cache data', async () => {
         const buffer = Buffer.from(await schemaRegistry.encode(registryId, payload))

diff --git a/src/SchemaRegistry.ts b/src/SchemaRegistry.ts
@@ -1,3 +1,5 @@
+import { Response } from 'mappersmith'
+
 import { encode, MAGIC_BYTE } from './encoder'
 import decode from './decoder'
 import { COMPATIBILITY, DEFAULT_SEPERATOR } from './constants'
@@ -24,8 +26,11 @@ const DEFAULT_OPTS = {
   separator: DEFAULT_SEPERATOR,
 }
 
+
 export default class SchemaRegistry {
   private api: SchemaRegistryAPIClient
+  private cacheMissRequests: { [key: number]: Promise<Response> } = {}
+
   public cache: Cache
 
   constructor({ auth, clientId, host, retry }: SchemaRegistryAPIClientArgs, options?: SchemaRegistryAPIClientOptions) {
@@ -84,11 +89,12 @@ export default class SchemaRegistry {
 
   public async getSchema(registryId: number): Promise<Schema> {
     const schema = this.cache.getSchema(registryId)
+
     if (schema) {
       return schema
     }
 
-    const response = await this.api.Schema.find({ id: registryId })
+    const response = await this.getSchemaOriginRequest(registryId)
     const foundSchema: { schema: string } = response.data()
     const rawSchema: Schema = JSON.parse(foundSchema.schema)
 
@@ -139,4 +145,19 @@ export default class SchemaRegistry {
 
     return id
   }
+
+  private getSchemaOriginRequest(registryId: number) {
+    // ensure that cache-misses result in a single origin request
+    if (this.cacheMissRequests[registryId]) {
+      return this.cacheMissRequests[registryId]
+    } else {
+      const request = this.api.Schema.find({ id: registryId }).finally(() => {
+        delete this.cacheMissRequests[registryId]
+      })
+
+      this.cacheMissRequests[registryId] = request
+
+      return request
+    }
+  }
 }