moving kafka to worker threads (#1963)

valhalla/jawn/src/index.ts

@@ -4,17 +4,16 @@ require("dotenv").config({
 
 import express, { NextFunction } from "express";
 import swaggerUi from "swagger-ui-express";
-import { Worker } from "worker_threads";
+import { tokenRouter } from "./lib/routers/tokenRouter";
 import { runLoopsOnce, runMainLoops } from "./mainLoops";
 import { authMiddleware } from "./middleware/auth";
+import { IS_RATE_LIMIT_ENABLED, limiter } from "./middleware/ratelimitter";
 import { RegisterRoutes as registerPrivateTSOARoutes } from "./tsoa-build/private/routes";
 import { RegisterRoutes as registerPublicTSOARoutes } from "./tsoa-build/public/routes";
 import * as publicSwaggerDoc from "./tsoa-build/public/swagger.json";
 import { initLogs } from "./utils/injectLogs";
 import { initSentry } from "./utils/injectSentry";
-import { IS_RATE_LIMIT_ENABLED, limiter } from "./middleware/ratelimitter";
-import { tokenRouter } from "./lib/routers/tokenRouter";
-import { consume, consumeDlq } from "./lib/clients/KafkaConsumer";
+import { startConsumers } from "./workers/consumerInterface";
 
 export const ENVIRONMENT: "production" | "development" = (process.env
   .VERCEL_ENV ?? "development") as any;
@@ -41,13 +40,12 @@ const app = express();
 
 const KAFKA_CREDS = JSON.parse(process.env.KAFKA_CREDS ?? "{}");
 const KAFKA_ENABLED = (KAFKA_CREDS?.KAFKA_ENABLED ?? "false") === "true";
+
 if (KAFKA_ENABLED) {
-  consume();
-  consumeDlq();
-}
-if (KAFKA_ENABLED) {
-  const worker = new Worker(`${__dirname}/workers/kafkaConsumer.js`);
-  worker.postMessage("start");
+  startConsumers({
+    dlqCount: 1,
+    normalCount: 1,
+  });
 }
 
 app.get("/healthcheck", (req, res) => {

valhalla/jawn/src/lib/clients/KafkaConsumer.ts

@@ -279,7 +279,9 @@ export const consumeDlq = async () => {
       heartbeat,
       commitOffsetsIfNecessary,
     }) => {
-      console.log(`Received batch with ${batch.messages.length} messages.`);
+      console.log(
+        `DLQ: Received batch with ${batch.messages.length} messages.`
+      );
       const maxMessages = 300;
       const miniBatches = createMiniBatches(batch.messages, maxMessages);
 
@@ -292,7 +294,7 @@ export const consumeDlq = async () => {
 
         const mappedMessages = mapDlqKafkaMessageToMessage(miniBatch);
         if (mappedMessages.error || !mappedMessages.data) {
-          console.error("Failed to map messages", mappedMessages.error);
+          console.error("DLQ: Failed to map messages", mappedMessages.error);
           return;
         }
 
@@ -305,7 +307,7 @@ export const consumeDlq = async () => {
           "request-response-logs-prod-dlq"
         );
         if (consumeResult.error) {
-          console.error("Failed to consume batch", consumeResult.error);
+          console.error("DLQ: Failed to consume batch", consumeResult.error);
           // TODO: Best way to handle this?
         }
 

valhalla/jawn/src/workers/consumerInterface.ts

@@ -0,0 +1,19 @@
+import { Worker } from "worker_threads";
+
+export function startConsumers({
+  normalCount,
+  dlqCount,
+}: {
+  normalCount: number;
+  dlqCount: number;
+}) {
+  for (let i = 0; i < normalCount; i++) {
+    const worker = new Worker(`${__dirname}/kafkaConsumer.js`);
+    worker.postMessage("start");
+  }
+
+  for (let i = 0; i < dlqCount; i++) {
+    const workerDlq = new Worker(`${__dirname}/kafkaConsumer.js`);
+    workerDlq.postMessage("start-dlq");
+  }
+}

valhalla/jawn/src/workers/kafkaConsumer.ts

@@ -1,9 +1,12 @@
 import { parentPort } from "worker_threads";
-import { consume } from "./../lib/clients/KafkaConsumer";
+import { consume, consumeDlq } from "./../lib/clients/KafkaConsumer";
 
 parentPort?.once("message", (message) => {
   if (message === "start") {
     console.log("Kafka consumer thread started!");
     consume();
+  } else if (message === "start-dlq") {
+    console.log("Kafka DLQ consumer thread started!");
+    consumeDlq();
   }
 });