Always using validated message when possible (#316)

* Always using validated message when possible

* Lint fix

Author: CarlosGamero

packages/kafka/lib/AbstractKafkaConsumer.ts

@@ -169,7 +169,7 @@ export abstract class AbstractKafkaConsumer<
       return this.commitMessage(message)
     }
 
-    const validatedMessage = parseResult.data
+    const validatedMessage = { ...message, value: parseResult.data }
 
     const requestContext = this.getRequestContext(message)
 
@@ -179,33 +179,29 @@ export abstract class AbstractKafkaConsumer<
       // exponential backoff -> 2^(retry-1)
       if (retries > 0) await setTimeout(Math.pow(2, retries - 1))
 
-      consumed = await this.tryToConsume(
-        { ...message, value: validatedMessage },
-        handler.handler,
-        requestContext,
-      )
+      consumed = await this.tryToConsume(validatedMessage, handler.handler, requestContext)
       if (consumed) break
 
       retries++
     } while (retries < MAX_IN_MEMORY_RETRIES)
 
     if (consumed) {
       this.handleMessageProcessed({
-        message: message,
+        message: validatedMessage,
         processingResult: { status: 'consumed' },
         messageProcessingStartTimestamp,
       })
     } else {
       this.handleMessageProcessed({
-        message: message,
+        message: validatedMessage,
         processingResult: { status: 'error', errorReason: 'handlerError' },
         messageProcessingStartTimestamp,
       })
     }
 
     this.transactionObservabilityManager?.stop(transactionId)
 
-    return this.commitMessage(message)
+    return this.commitMessage(validatedMessage)
   }
 
   private async tryToConsume<MessageValue extends object>(

packages/kafka/package.json

@@ -1,6 +1,6 @@
 {
     "name": "@message-queue-toolkit/kafka",
-    "version": "0.3.1",
+    "version": "0.3.2",
     "engines": {
         "node": ">= 22.14.0"
     },