Add HTTP error handling (#127)

kravets-levko · nithinkdb · web-flow · commit 15e5fa40225f · 2023-04-25T17:50:16.000+03:00
* Add error handling (#102) Signed-off-by: Levko Kravets <levko.ne@gmail.com> Signed-off-by: nithinkdb <nithin.krishnamurthi@databricks.com> * Refactoring: Move error handler to BaseCommand Signed-off-by: Levko Kravets <levko.ne@gmail.com> * Refactoring: refine error handling code Signed-off-by: Levko Kravets <levko.ne@gmail.com> * Add delay between retries + make it configurable through constants Signed-off-by: Levko Kravets <levko.ne@gmail.com> * Add tests Signed-off-by: Levko Kravets <levko.ne@gmail.com> * Fix test name Signed-off-by: Levko Kravets <levko.ne@gmail.com> * CR1 Signed-off-by: Levko Kravets <levko.ne@gmail.com> --------- Signed-off-by: nithinkdb <nithin.krishnamurthi@databricks.com> Signed-off-by: Levko Kravets <levko.ne@gmail.com> Co-authored-by: Nithin Krishnamurthi <110488533+nithinkdb@users.noreply.github.com>
diff --git a/.eslintrc b/.eslintrc
@@ -13,7 +13,8 @@
         "no-param-reassign": "off",
         "no-bitwise": "off",
         "@typescript-eslint/no-throw-literal": "off",
-        "no-restricted-syntax": "off"
+        "no-restricted-syntax": "off",
+        "no-case-declarations": "off"
       }
     }
   ]
diff --git a/lib/globalConfig.ts b/lib/globalConfig.ts
@@ -1,9 +1,19 @@
 interface GlobalConfig {
   arrowEnabled?: boolean;
   useArrowNativeTypes?: boolean;
+
+  retryMaxAttempts: number;
+  retriesTimeout: number; // in milliseconds
+  retryDelayMin: number; // in milliseconds
+  retryDelayMax: number; // in milliseconds
 }
 
 export default {
   arrowEnabled: true,
   useArrowNativeTypes: true,
+
+  retryMaxAttempts: 30,
+  retriesTimeout: 900 * 1000,
+  retryDelayMin: 1 * 1000,
+  retryDelayMax: 60 * 1000,
 } satisfies GlobalConfig;
diff --git a/lib/hive/Commands/BaseCommand.ts b/lib/hive/Commands/BaseCommand.ts
@@ -1,5 +1,23 @@
+import { Thrift } from 'thrift';
 import TCLIService from '../../../thrift/TCLIService';
 import HiveDriverError from '../../errors/HiveDriverError';
+import globalConfig from '../../globalConfig';
+
+interface CommandExecutionInfo {
+  startTime: number; // in milliseconds
+  attempt: number;
+}
+
+function getRetryDelay(attempt: number): number {
+  const scale = Math.max(1, 1.5 ** (attempt - 1)); // ensure scale >= 1
+  return Math.min(globalConfig.retryDelayMin * scale, globalConfig.retryDelayMax);
+}
+
+function delay(milliseconds: number): Promise<void> {
+  return new Promise<void>((resolve) => {
+    setTimeout(() => resolve(), milliseconds);
+  });
+}
 
 export default abstract class BaseCommand {
   protected client: TCLIService.Client;
@@ -8,13 +26,71 @@ export default abstract class BaseCommand {
     this.client = client;
   }
 
-  executeCommand<Response>(request: object, command: Function | void): Promise<Response> {
-    return new Promise((resolve, reject) => {
-      if (typeof command !== 'function') {
-        reject(new HiveDriverError('Hive driver: the operation does not exist, try to choose another Thrift file.'));
-        return;
+  protected executeCommand<Response>(request: object, command: Function | void): Promise<Response> {
+    return this.invokeWithErrorHandling<Response>(request, command, { startTime: Date.now(), attempt: 0 });
+  }
+
+  private async invokeWithErrorHandling<Response>(
+    request: object,
+    command: Function | void,
+    info: CommandExecutionInfo,
+  ): Promise<Response> {
+    try {
+      return await this.invokeCommand<Response>(request, command);
+    } catch (error) {
+      if (error instanceof Thrift.TApplicationException) {
+        if ('statusCode' in error) {
+          switch (error.statusCode) {
+            // On these status codes it's safe to retry the request. However,
+            // both error codes mean that server is overwhelmed or even down.
+            // Therefore, we need to add some delay between attempts so
+            // server can recover and more likely handle next request
+            case 429: // Too Many Requests
+            case 503: // Service Unavailable
+              info.attempt += 1;
+
+              // Delay interval depends on current attempt - the more attempts we do
+              // the longer the interval will be
+              // TODO: Respect `Retry-After` header (PECO-729)
+              const retryDelay = getRetryDelay(info.attempt);
+
+              const attemptsExceeded = info.attempt >= globalConfig.retryMaxAttempts;
+              if (attemptsExceeded) {
+                throw new HiveDriverError(
+                  `Hive driver: ${error.statusCode} when connecting to resource. Max retry count exceeded.`,
+                );
+              }
+
+              const timeoutExceeded = Date.now() - info.startTime + retryDelay >= globalConfig.retriesTimeout;
+              if (timeoutExceeded) {
+                throw new HiveDriverError(
+                  `Hive driver: ${error.statusCode} when connecting to resource. Retry timeout exceeded.`,
+                );
+              }
+
+              await delay(retryDelay);
+              return this.invokeWithErrorHandling(request, command, info);
+
+            // TODO: Here we should handle other error types (see PECO-730)
+
+            // no default
+          }
+        }
       }
 
+      // Re-throw error we didn't handle
+      throw error;
+    }
+  }
+
+  private invokeCommand<Response>(request: object, command: Function | void): Promise<Response> {
+    if (typeof command !== 'function') {
+      return Promise.reject(
+        new HiveDriverError('Hive driver: the operation does not exist, try to choose another Thrift file.'),
+      );
+    }
+
+    return new Promise((resolve, reject) => {
       try {
         command.call(this.client, request, (err: Error, response: Response) => {
           if (err) {
diff --git a/tests/unit/hive/commands/BaseCommand.test.js b/tests/unit/hive/commands/BaseCommand.test.js
@@ -0,0 +1,230 @@
+const { expect, AssertionError } = require('chai');
+const { Thrift } = require('thrift');
+const HiveDriverError = require('../../../../dist/errors/HiveDriverError').default;
+const BaseCommand = require('../../../../dist/hive/Commands/BaseCommand').default;
+const globalConfig = require('../../../../dist/globalConfig').default;
+
+const savedGlobalConfig = { ...globalConfig };
+
+class ThriftClientMock {
+  constructor(methodHandler) {
+    this.methodHandler = methodHandler;
+  }
+
+  CustomMethod(request, callback) {
+    try {
+      const response = this.methodHandler();
+      return callback(undefined, response !== undefined ? response : ThriftClientMock.defaultResponse);
+    } catch (error) {
+      return callback(error);
+    }
+  }
+}
+
+ThriftClientMock.defaultResponse = {
+  status: { statusCode: 0 },
+};
+
+class CustomCommand extends BaseCommand {
+  execute(request) {
+    return this.executeCommand(request, this.client.CustomMethod);
+  }
+}
+
+describe('BaseCommand', () => {
+  afterEach(() => {
+    Object.assign(globalConfig, savedGlobalConfig);
+  });
+
+  it('should fail if trying to invoke non-existing command', async () => {
+    const command = new CustomCommand({});
+
+    try {
+      await command.execute();
+      expect.fail('It should throw an error');
+    } catch (error) {
+      if (error instanceof AssertionError) {
+        throw error;
+      }
+      expect(error).to.be.instanceof(HiveDriverError);
+      expect(error.message).to.contain('the operation does not exist');
+    }
+  });
+
+  it('should handle exceptions thrown by command', async () => {
+    const errorMessage = 'Unexpected error';
+
+    const command = new CustomCommand({
+      CustomMethod() {
+        throw new Error(errorMessage);
+      },
+    });
+
+    try {
+      await command.execute();
+      expect.fail('It should throw an error');
+    } catch (error) {
+      if (error instanceof AssertionError) {
+        throw error;
+      }
+      expect(error).to.be.instanceof(Error);
+      expect(error.message).to.contain(errorMessage);
+    }
+  });
+
+  [429, 503].forEach((statusCode) => {
+    describe(`HTTP ${statusCode} error`, () => {
+      it('should fail on max retry attempts exceeded', async () => {
+        globalConfig.retriesTimeout = 200; // ms
+        globalConfig.retryDelayMin = 5; // ms
+        globalConfig.retryDelayMax = 20; // ms
+        globalConfig.retryMaxAttempts = 3;
+
+        let methodCallCount = 0;
+        const command = new CustomCommand(
+          new ThriftClientMock(() => {
+            methodCallCount += 1;
+            const error = new Thrift.TApplicationException();
+            error.statusCode = statusCode;
+            throw error;
+          }),
+        );
+
+        try {
+          await command.execute();
+          expect.fail('It should throw an error');
+        } catch (error) {
+          if (error instanceof AssertionError) {
+            throw error;
+          }
+          expect(error).to.be.instanceof(HiveDriverError);
+          expect(error.message).to.contain(`${statusCode} when connecting to resource`);
+          expect(error.message).to.contain('Max retry count exceeded');
+          expect(methodCallCount).to.equal(globalConfig.retryMaxAttempts);
+        }
+      });
+
+      it('should fail on retry timeout exceeded', async () => {
+        globalConfig.retriesTimeout = 200; // ms
+        globalConfig.retryDelayMin = 5; // ms
+        globalConfig.retryDelayMax = 20; // ms
+        globalConfig.retryMaxAttempts = 50;
+
+        let methodCallCount = 0;
+        const command = new CustomCommand(
+          new ThriftClientMock(() => {
+            methodCallCount += 1;
+            const error = new Thrift.TApplicationException();
+            error.statusCode = statusCode;
+            throw error;
+          }),
+        );
+
+        try {
+          await command.execute();
+          expect.fail('It should throw an error');
+        } catch (error) {
+          if (error instanceof AssertionError) {
+            throw error;
+          }
+          expect(error).to.be.instanceof(HiveDriverError);
+          expect(error.message).to.contain(`${statusCode} when connecting to resource`);
+          expect(error.message).to.contain('Retry timeout exceeded');
+          // We set pretty low intervals/timeouts to make this test pass faster, but it also means
+          // that it's harder to predict how much times command will be invoked. So we check that
+          // it is within some meaningful range to reduce false-negatives
+          expect(methodCallCount).to.be.within(10, 20);
+        }
+      });
+
+      it('should succeed after few attempts', async () => {
+        globalConfig.retriesTimeout = 200; // ms
+        globalConfig.retryDelayMin = 5; // ms
+        globalConfig.retryDelayMax = 20; // ms
+        globalConfig.retryMaxAttempts = 5;
+
+        let methodCallCount = 0;
+        const command = new CustomCommand(
+          new ThriftClientMock(() => {
+            methodCallCount += 1;
+            if (methodCallCount <= 3) {
+              const error = new Thrift.TApplicationException();
+              error.statusCode = statusCode;
+              throw error;
+            }
+            return ThriftClientMock.defaultResponse;
+          }),
+        );
+
+        const response = await command.execute();
+        expect(response).to.deep.equal(ThriftClientMock.defaultResponse);
+        expect(methodCallCount).to.equal(4); // 3 failed attempts + 1 succeeded
+      });
+    });
+  });
+
+  it(`should re-throw unrecognized HTTP errors`, async () => {
+    const errorMessage = 'Unrecognized HTTP error';
+
+    const command = new CustomCommand(
+      new ThriftClientMock(() => {
+        const error = new Thrift.TApplicationException(undefined, errorMessage);
+        error.statusCode = 500;
+        throw error;
+      }),
+    );
+
+    try {
+      await command.execute();
+      expect.fail('It should throw an error');
+    } catch (error) {
+      if (error instanceof AssertionError) {
+        throw error;
+      }
+      expect(error).to.be.instanceof(Thrift.TApplicationException);
+      expect(error.message).to.contain(errorMessage);
+    }
+  });
+
+  it(`should re-throw unrecognized Thrift errors`, async () => {
+    const errorMessage = 'Unrecognized HTTP error';
+
+    const command = new CustomCommand(
+      new ThriftClientMock(() => {
+        throw new Thrift.TApplicationException(undefined, errorMessage);
+      }),
+    );
+
+    try {
+      await command.execute();
+      expect.fail('It should throw an error');
+    } catch (error) {
+      if (error instanceof AssertionError) {
+        throw error;
+      }
+      expect(error).to.be.instanceof(Thrift.TApplicationException);
+      expect(error.message).to.contain(errorMessage);
+    }
+  });
+
+  it(`should re-throw unrecognized errors`, async () => {
+    const errorMessage = 'Unrecognized error';
+
+    const command = new CustomCommand(
+      new ThriftClientMock(() => {
+        throw new Error(errorMessage);
+      }),
+    );
+
+    try {
+      await command.execute();
+      expect.fail('It should throw an error');
+    } catch (error) {
+      if (error instanceof AssertionError) {
+        throw error;
+      }
+      expect(error).to.be.instanceof(Error);
+      expect(error.message).to.contain(errorMessage);
+    }
+  });
+});

Original file line number	Diff line number	Diff line change
`@@ -13,7 +13,8 @@`
`13`	`13`	`"no-param-reassign": "off",`
`14`	`14`	`"no-bitwise": "off",`
`15`	`15`	`"@typescript-eslint/no-throw-literal": "off",`
`16`		`- "no-restricted-syntax": "off"`
	`16`	`+ "no-restricted-syntax": "off",`
	`17`	`+ "no-case-declarations": "off"`
`17`	`18`	`}`
`18`	`19`	`}`
`19`	`20`	`]`