fix(kafka): add status code & batched value in response structure (#2684

) * fix: add status code & batched value in response structure * fix: check for erroneous events * chore: improve test coverage
rudderlabs · Oct 5, 2023 · 99f39f5 · 99f39f5
1 parent e37d693
commit 99f39f5
Show file tree

Hide file tree

Showing 4 changed files with 1,416 additions and 1,295 deletions.
diff --git a/src/v0/destinations/kafka/transform.js b/src/v0/destinations/kafka/transform.js
@@ -5,6 +5,8 @@ const {
   getIntegrationsObj,
   getHashFromArray,
   removeUndefinedAndNullValues,
+  getSuccessRespEvents,
+  getErrorRespEvents,
 } = require('../../util');
 // const { InstrumentationError } = require("../../util/errorTypes");
 
@@ -37,6 +39,10 @@ const filterConfigTopics = (message, destination) => {
 
 const batch = (destEvents) => {
   const respList = [];
+  if (!Array.isArray(destEvents) || destEvents.length <= 0) {
+    const respEvents = getErrorRespEvents(null, 400, 'Invalid event array');
+    return [respEvents];
+  }
 
   // Grouping the events by topic
   const groupedEvents = groupBy(destEvents, (event) => event.message.topic);
@@ -52,9 +58,10 @@ const batch = (destEvents) => {
       metadata: events.map((event) => event.metadata),
       destination: events[0].destination,
     };
-    respList.push(response);
+    respList.push(
+      getSuccessRespEvents(response.batchedRequest, response.metadata, response.destination, true),
+    );
   }
-
   return respList;
 };