Add retry logic to RestService (#165)

Author: rayokota

package-lock.json

@@ -44,6 +44,7 @@
     "async-mutex": "^0.5.0",
     "avsc": "^5.7.7",
     "axios": "^1.7.3",
+    "axios-retry": "^4.5.0",
     "json-stringify-deterministic": "^1.0.12",
     "jsonata": "^2.0.5",
     "lru-cache": "^10.4.3",

schemaregistry/package.json

@@ -1,6 +1,7 @@
 import axios, { AxiosInstance, AxiosRequestConfig, AxiosResponse, CreateAxiosDefaults } from 'axios';
 import { OAuthClient } from './oauth/oauth-client';
 import { RestError } from './rest-error';
+import axiosRetry from "axios-retry";
 /*
  * Confluent-Schema-Registry-TypeScript - Node.js wrapper for Confluent Schema Registry
  *
@@ -33,7 +34,6 @@ export interface BearerAuthCredentials {
   identityPoolId?: string,
 }
 
-//TODO: Consider retry policy, may need additional libraries on top of Axios
 export interface ClientConfig {
   baseURLs: string[],
   cacheCapacity?: number,
@@ -42,6 +42,9 @@ export interface ClientConfig {
   createAxiosDefaults?: CreateAxiosDefaults,
   basicAuthCredentials?: BasicAuthCredentials,
   bearerAuthCredentials?: BearerAuthCredentials,
+  maxRetries?: number,
+  retriesWaitMs?: number,
+  retriesMaxWaitMs?: number,
 }
 
 const toBase64 = (str: string): string => Buffer.from(str).toString('base64');
@@ -53,19 +56,38 @@ export class RestService {
   private oauthBearer: boolean = false;
 
   constructor(baseURLs: string[], isForward?: boolean, axiosDefaults?: CreateAxiosDefaults,
-    basicAuthCredentials?: BasicAuthCredentials, bearerAuthCredentials?: BearerAuthCredentials) {
+              basicAuthCredentials?: BasicAuthCredentials, bearerAuthCredentials?: BearerAuthCredentials,
+              maxRetries?: number, retriesWaitMs?: number, retriesMaxWaitMs?: number) {
     this.client = axios.create(axiosDefaults);
+    axiosRetry(this.client, {
+      retries: maxRetries ?? 2,
+      retryDelay: (retryCount) => {
+        return this.fullJitter(retriesWaitMs ?? 1000, retriesMaxWaitMs ?? 20000, retryCount - 1)
+      },
+      retryCondition: (error) => {
+        return this.isRetriable(error.response?.status ?? 0);
+      }
+    });
     this.baseURLs = baseURLs;
 
     if (isForward) {
       this.setHeaders({ 'X-Forward': 'true' });
     }
     this.setHeaders({ 'Content-Type': 'application/vnd.schemaregistry.v1+json' });
-    
+
     this.handleBasicAuth(basicAuthCredentials);
     this.handleBearerAuth(bearerAuthCredentials);
   }
 
+  isRetriable(statusCode: number): boolean {
+    return statusCode == 408 || statusCode == 429
+      || statusCode == 500 || statusCode == 502 || statusCode == 503 || statusCode == 504;
+  }
+
+  fullJitter(baseDelayMs: number, maxDelayMs: number, retriesAttempted: number): number {
+    return Math.random() * Math.min(maxDelayMs, baseDelayMs * 2 ** retriesAttempted)
+  }
+
   handleBasicAuth(basicAuthCredentials?: BasicAuthCredentials): void {
     if (basicAuthCredentials) {
       switch (basicAuthCredentials.credentialsSource) {

schemaregistry/rest-service.ts

@@ -71,7 +71,8 @@ class DekRegistryClient implements DekClient {
 
 
     this.restService = new RestService(config.baseURLs, config.isForward, config.createAxiosDefaults,
-      config.basicAuthCredentials, config.bearerAuthCredentials);
+      config.basicAuthCredentials, config.bearerAuthCredentials,
+      config.maxRetries, config.retriesWaitMs, config.retriesMaxWaitMs);
     this.kekCache = new LRUCache<string, Kek>(cacheOptions);
     this.dekCache = new LRUCache<string, Dek>(cacheOptions);
     this.kekMutex = new Mutex();

schemaregistry/rules/encryption/dekregistry/dekregistry-client.ts

@@ -202,7 +202,8 @@ export class SchemaRegistryClient implements Client {
     };
 
     this.restService = new RestService(config.baseURLs, config.isForward, config.createAxiosDefaults,
-      config.basicAuthCredentials, config.bearerAuthCredentials);
+      config.basicAuthCredentials, config.bearerAuthCredentials,
+      config.maxRetries, config.retriesWaitMs, config.retriesMaxWaitMs);
 
     this.schemaToIdCache = new LRUCache(cacheOptions);
     this.idToSchemaInfoCache = new LRUCache(cacheOptions);