feat(api): implement batch processing for Kafka messages and bulk insert into ClickHouse

Author: Prateet-Github

api/worker/index.ts

@@ -1,6 +1,7 @@
 import { Kafka } from 'kafkajs';
 import { env } from '../config/env.js';
 import { clickhouse, connectClickHouse } from '../lib/clickhouse.js';
+import crypto from 'node:crypto';
 
 const kafka = new Kafka({
   clientId: 'logstream-worker',
@@ -10,44 +11,60 @@ const kafka = new Kafka({
 const consumer = kafka.consumer({ groupId: 'log-processing-group-v3' });
 
 const run = async () => {
-  // connect to infrastructure
+
   await connectClickHouse();
   await consumer.connect();
   console.log('Worker connected to Kafka');
 
   await consumer.subscribe({ topic: 'app-logs', fromBeginning: true });
 
-  // process messages
+  // swtiched to batch processing from single message processing for efficiency
   await consumer.run({
-    eachMessage: async ({ topic, partition, message }) => {
-      const logString = message.value?.toString();
+    eachBatchAutoResolve: true,
+    eachBatch: async ({ batch, resolveOffset, heartbeat }) => {
+      const messages = batch.messages;
 
-      if (!logString) return;
+      if (messages.length === 0) return;
 
+      console.log(`Processing batch of ${messages.length} messages...`);
 
-      try {
-        const logData = JSON.parse(logString);
+      // Transform Kafka Messages to ClickHouse Rows
+      const rows = messages.map((message) => {
+        const logString = message.value?.toString();
+        if (!logString) return null;
+
+        try {
+          const logData = JSON.parse(logString);
+          return {
+            id: logData.id || crypto.randomUUID(),
+            service: logData.service,
+            level: logData.level,
+            message: logData.message,
+            meta: JSON.stringify(logData.meta || {}),
+            timestamp: new Date(logData.timestamp).getTime(),
+          };
+        } catch (err) {
+          return null; 
+        }
+      }).filter(Boolean);
 
-        // insert into clickHouse
+      if (rows.length === 0) return;
+
+      try {
+        // bulk Insert into ClickHouse 
         await clickhouse.insert({
           table: 'app_logs',
-          values: [
-            {
-              id: logData.id || crypto.randomUUID(), // Fallback ID
-              service: logData.service,
-              level: logData.level,
-              message: logData.message,
-              meta: JSON.stringify(logData.meta || {}),
-              timestamp: new Date(logData.timestamp).getTime(),
-            },
-          ],
+          values: rows,
           format: 'JSONEachRow',
         });
 
-        console.log(`Inserted log from ${logData.service}`);
+        // mark success
+        await heartbeat(); 
+        
+        console.log(`committed ${rows.length} logs to ClickHouse`);
 
       } catch (err) {
-        console.error('Error processing message:', err);
+        console.error('Batch Insert Failed:', err);
       }
     },
   });