refactor(kafka): improve tests & error handling (#4262)

Author: dreamorosi

packages/kafka/src/deserializer/protobuf.ts

@@ -40,7 +40,7 @@ const deserialize = <T>(
     }
   } catch (error) {
     throw new KafkaConsumerDeserializationError(
-      `Failed to deserialize Protobuf message: ${error}, message: ${data}, messageType: ${messageType}`
+      `Failed to deserialize Protobuf message: ${error}, message: ${data}, messageType: ${JSON.stringify(messageType)}`
     );
   }
 
@@ -64,7 +64,7 @@ const deserialize = <T>(
       return decoded;
     } catch {
       throw new KafkaConsumerDeserializationError(
-        `Failed to deserialize Protobuf message: ${error}, message: ${data}, messageType: ${messageType}`
+        `Failed to deserialize Protobuf message: ${error}, message: ${data}, messageType: ${JSON.stringify(messageType)}`
       );
     }
   }

packages/kafka/tests/unit/consumer.test.ts

@@ -141,13 +141,15 @@ describe('Kafka consumer', () => {
       // Prepare
       const handler = kafkaConsumer(
         async (event) => {
+          const results = [];
           for (const record of event.records) {
             try {
-              return record.value;
+              results.push(record.value);
             } catch (error) {
               return error;
             }
           }
+          return results;
         },
         {
           // @ts-expect-error - testing missing schemaStr
@@ -241,14 +243,16 @@ describe('Kafka consumer', () => {
       // Prepare
       const handler = kafkaConsumer(
         async (event) => {
+          const results = [];
           for (const record of event.records) {
             try {
               const { value, key } = record;
-              return [value, key];
+              results.push([value, key]);
             } catch (error) {
               return error;
             }
           }
+          return results;
         },
         {
           value: {