Merge pull request #13 from devit-tel/feature/add-batch-timeout

Feature/add batch timeout

Author: NV4RE

package-lock.json

@@ -20,13 +20,15 @@
   "dependencies": {
     "@melonade/melonade-declaration": "^0.19.2",
     "axios": "^0.20.0",
-    "node-rdkafka": "^2.9.1",
+    "node-rdkafka": "^2.10.0",
+    "promise-timeout": "^1.3.0",
     "ramda": "^0.26.1",
     "tslib": "^2.0.1"
   },
   "devDependencies": {
     "@types/jest": "^24.0.23",
     "@types/node": "^10.17.5",
+    "@types/promise-timeout": "^1.3.0",
     "@types/ramda": "^0.26.33",
     "jest": "^26.4.2",
     "prettier": "^2.0.5",

package.json

@@ -1,13 +1,13 @@
 import { State, Task } from '@melonade/melonade-declaration';
 import { SyncWorker, TaskStates } from '..';
 
-const kafkaServers = process.env['MELONADE_KAFKA_SERVERS'];
-const namespace = process.env['MELONADE_NAMESPACE'];
-const processManagerUrl =
-  process.env['MELONADE_PROCESS_MANAGER_URL'] || 'http://localhost:8081';
+const kafkaServers = 'localhost:29092';
+const namespace = 'docker-compose';
+const processManagerUrl = 'http://localhost:8081';
 
 const sleep = (ms: number) => new Promise((res) => setTimeout(res, ms));
 
+// tslint:disable-next-line: no-for-in
 for (const forkID in new Array(1).fill(null)) {
   for (const workerId of [1, 2, 3]) {
     const worker = new SyncWorker(

src/example/syncWorker.ts

@@ -1,28 +1,25 @@
 import { State, Task } from '@melonade/melonade-declaration';
 import { Worker } from '..';
 
-const kafkaServers = process.env['MELONADE_KAFKA_SERVERS'];
-const namespace = process.env['MELONADE_NAMESPACE'];
+const kafkaServers = 'localhost:29092';
+const namespace = 'docker-compose';
 
+const sleep = (ms: number) => new Promise((res) => setTimeout(res, ms));
+
+// tslint:disable-next-line: no-for-in
 for (const forkID in new Array(1).fill(null)) {
-  for (const workerId of [1, 2, 3]) {
+  for (const workerId of [4]) {
     const worker = new Worker(
       // task name
       `t${workerId}`,
 
       // process task
-      (task, _logger, _isTimeOut, updateTask) => {
-        setTimeout(() => {
-          updateTask(task, {
-            status: State.TaskStates.Completed,
-          });
-
-          console.log(`Async Completed ${task.taskName}`);
-        }, 5000);
-
-        console.log(`Processing ${task.taskName}`);
+      async (task, _logger, _isTimeOut) => {
+        console.log(`Processing ${task.taskName}: ${task.taskId}`);
+        await sleep(5 * 1000);
+        console.log(`Processed ${task.taskName}:  ${task.taskId}`);
         return {
-          status: State.TaskStates.Inprogress,
+          status: State.TaskStates.Completed,
         };
       },
 

src/example/worker.ts

@@ -7,6 +7,8 @@ import {
   LibrdKafkaError,
   Message,
 } from 'node-rdkafka';
+import { timeout, TimeoutError } from 'promise-timeout';
+import * as R from 'ramda';
 import { jsonTryParse } from './utils/common';
 import {
   isTaskTimeout,
@@ -25,6 +27,7 @@ export interface ISyncWorkerConfig {
   autoStart?: boolean;
   latencyCompensationMs?: number;
   trackingRunningTasks?: boolean;
+  batchTimeoutMs?: number;
 }
 
 export interface ISyncUpdateTask {
@@ -39,6 +42,7 @@ const DEFAULT_WORKER_CONFIG = {
   autoStart: true,
   latencyCompensationMs: 50,
   trackingRunningTasks: false,
+  batchTimeoutMs: 10 * 60 * 1000, // 10 mins
 } as ISyncWorkerConfig;
 
 // Maybe use kafka streamAPI
@@ -57,7 +61,7 @@ export class SyncWorker extends EventEmitter {
     isTimeout: boolean,
   ) => void | Promise<void>;
   private runningTasks: {
-    [taskId: string]: Task.ITask;
+    [taskId: string]: Task.ITask | string;
   } = {};
   private tasksName: string | string[];
 
@@ -98,6 +102,10 @@ export class SyncWorker extends EventEmitter {
         'bootstrap.servers': workerConfig.kafkaServers,
         'group.id': `melonade-${this.workerConfig.namespace}-client-${tn}`,
         'enable.auto.commit': false,
+        'max.poll.interval.ms': Math.max(
+          300000,
+          this.workerConfig.batchTimeoutMs * 5,
+        ),
         ...kafkaConfig,
       },
       { 'auto.offset.reset': 'latest' },
@@ -126,13 +134,14 @@ export class SyncWorker extends EventEmitter {
     this.consumer.connect();
 
     process.once('SIGTERM', () => {
+      console.log(`${this.tasksName}: unsubscribed`);
       this.consumer.unsubscribe();
     });
   }
 
   get health(): {
     consumer: 'connected' | 'disconnected';
-    tasks: { [taskId: string]: Task.ITask };
+    tasks: { [taskId: string]: Task.ITask | string };
   } {
     return {
       consumer: this.consumer.isConnected() ? 'connected' : 'disconnected',
@@ -143,12 +152,13 @@ export class SyncWorker extends EventEmitter {
   consume = (
     messageNumber: number = this.workerConfig.maximumPollingTasks,
   ): Promise<Task.ITask[]> => {
-    return new Promise((resolve: Function, reject: Function) => {
+    return new Promise((resolve: Function) => {
       this.consumer.consume(
         messageNumber,
         (error: LibrdKafkaError, messages: Message[]) => {
           if (error) {
-            setTimeout(() => reject(error), 1000);
+            console.log(`${this.tasksName}: consume error`, error);
+            setTimeout(() => resolve([]), 1000);
           } else {
             resolve(
               messages.map((message: Kafka.kafkaConsumerMessage) =>
@@ -181,7 +191,12 @@ export class SyncWorker extends EventEmitter {
   };
 
   commit = () => {
-    return this.consumer.commit();
+    try {
+      // @ts-ignore
+      this.consumer.commitSync();
+    } catch (error) {
+      console.log(`${this.tasksName}: commit error`, error);
+    }
   };
 
   private dispatchTask = async (task: Task.ITask, isTimeout: boolean) => {
@@ -207,31 +222,45 @@ export class SyncWorker extends EventEmitter {
 
     if (this.workerConfig.trackingRunningTasks) {
       this.runningTasks[task.taskId] = task;
+    } else {
+      this.runningTasks[task.taskId] = task.taskId;
     }
 
     try {
       await this.dispatchTask(task, isTimeout);
     } catch (error) {
       console.warn(this.tasksName, error);
     } finally {
-      if (this.workerConfig.trackingRunningTasks) {
-        delete this.runningTasks[task.taskId];
-      }
+      delete this.runningTasks[task.taskId];
     }
   };
 
   private poll = async () => {
     // https://github.com/nodejs/node/issues/6673
     while (this.isSubscribed) {
-      try {
-        const tasks = await this.consume();
-        if (tasks.length > 0) {
-          await Promise.all(tasks.map(this.processTask));
+      const tasks = await this.consume();
+      if (tasks.length > 0) {
+        try {
+          if (this.workerConfig.batchTimeoutMs > 0) {
+            await timeout(
+              Promise.all(tasks.map(this.processTask)),
+              this.workerConfig.batchTimeoutMs,
+            );
+          } else {
+            await Promise.all(tasks.map(this.processTask));
+          }
+        } catch (error) {
+          if (error instanceof TimeoutError) {
+            console.log(
+              `${this.tasksName}: batch timeout`,
+              R.keys(this.runningTasks),
+            );
+          } else {
+            console.log(this.tasksName, 'process error', error);
+          }
+        } finally {
           this.commit();
         }
-      } catch (err) {
-        // In case of consume error
-        console.log(this.tasksName, err);
       }
     }