Repository: nestjs/nest
Pull request #9996 (2022-07-21 05:50:41)

fix(microservices): prevent kafka parser from modifying received message
## PR Checklist
Please check if your PR fulfills the following requirements:

- [x] The commit message follows our guidelines: https://github.com/nestjs/nest/blob/master/CONTRIBUTING.md
- [x] Tests for the changes have been added (for bug fixes / features)
- [ ] Docs have been added / updated (for bug fixes / features)


## PR Type
What kind of change does this PR introduce?

<!-- Please check the one that applies to this PR using "x". -->
- [x] Bugfix
- [ ] Feature
- [ ] Code style update (formatting, local variables)
- [ ] Refactoring (no functional changes, no api changes)
- [ ] Build related changes
- [ ] CI related changes
- [ ] Other... Please describe:

## What is the current behavior?
Kafka parser modifies messages during parsing. If a message handler fails and message is retried by KafkaJS, then the already parsed message is parsed again and the value ends up being the string `'[object Object]'`. 

Issue Number: 9956


## What is the new behavior?
Kafka parser duplicates message objects and puts parsed data into these newly created objects. So retries are executed with the same data.

## Does this PR introduce a breaking change?
- [ ] Yes
- [x] No


## Other information

fix: #9956 

coveralls: ## Pull Request Test Coverage Report for [Build 2a916288-522f-47f2-b435-d042adfda12e](https://coveralls.io/builds/51041060)

* **6** of **6**   **(100.0%)**  changed or added relevant lines in **1** file are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage increased (+**0.0009%**) to **94.024%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/51041060/badge)](https://coveralls.io/builds/51041060) |
| :-- | --: |
| Change from base [Build b0497842-ca88-4578-a1ef-e27581f2ef4f](https://coveralls.io/builds/51014361): |  0.0009% |
| Covered Lines: | 6057 |
| Relevant Lines: | 6442 |

---
##### 💛  - [Coveralls](https://coveralls.io)

Tony133: Hi @jDzames,  `CompressionCodecs ` in `kafka.interface.ts`, has been fixed in this PR https://github.com/nestjs/nest/pull/9991


Patch:
diff --git a/packages/microservices/helpers/kafka-parser.ts b/packages/microservices/helpers/kafka-parser.ts
index 781217ff8..61056e7c1 100644
--- a/packages/microservices/helpers/kafka-parser.ts
+++ b/packages/microservices/helpers/kafka-parser.ts
@@ -9,22 +9,28 @@ export class KafkaParser {
   }
 
   public parse<T = any>(data: any): T {
+    // Clone object to as modifying the original one would break KafkaJS retries
+    const result = {
+      ...data,
+      headers: { ...data.headers },
+    };
+
     if (!this.keepBinary) {
-      data.value = this.decode(data.value);
+      result.value = this.decode(data.value);
     }
 
     if (!isNil(data.key)) {
-      data.key = this.decode(data.key);
+      result.key = this.decode(data.key);
     }
     if (!isNil(data.headers)) {
       const decodeHeaderByKey = (key: string) => {
-        data.headers[key] = this.decode(data.headers[key]);
+        result.headers[key] = this.decode(data.headers[key]);
       };
       Object.keys(data.headers).forEach(decodeHeaderByKey);
     } else {
-      data.headers = {};
+      result.headers = {};
     }
-    return data;
+    return result;
   }
 
   public decode(value: Buffer): object | string | null | Buffer {


Test patch:
diff --git a/packages/microservices/test/client/client-kafka.spec.ts b/packages/microservices/test/client/client-kafka.spec.ts
index 3770baea9..39743da52 100644
--- a/packages/microservices/test/client/client-kafka.spec.ts
+++ b/packages/microservices/test/client/client-kafka.spec.ts
@@ -76,7 +76,7 @@ describe('ClientKafka', () => {
       message,
       {
         size: 0,
-        value: { test: true },
+        value: Buffer.from(JSON.stringify({ test: true })),
       },
     ),
     heartbeat,
@@ -488,7 +488,7 @@ describe('ClientKafka', () => {
         expect(
           callback.calledWith({
             isDisposed: true,
-            response: payloadDisposed.message.value,
+            response: deserializedPayloadDisposed.message.value,
             err: undefined,
           }),
         ).to.be.true;
diff --git a/packages/microservices/test/helpers/kafka-parser.spec.ts b/packages/microservices/test/helpers/kafka-parser.spec.ts
index 91c698d1b..e4f8babcd 100644
--- a/packages/microservices/test/helpers/kafka-parser.spec.ts
+++ b/packages/microservices/test/helpers/kafka-parser.spec.ts
@@ -126,5 +126,35 @@ describe('KafkaParser', () => {
         },
       });
     });
+
+    it('parse message multiple times (simulate retry)', () => {
+      const message = {
+        headers: {
+          [KafkaHeaders.CORRELATION_ID]: Buffer.from('correlation-id'),
+        },
+        value: Buffer.from(JSON.stringify({ prop: 'value' })),
+        key: Buffer.from('1'),
+      };
+      const expectedParsedMessage = {
+        key: '1',
+        value: {
+          prop: 'value',
+        },
+        headers: {
+          [KafkaHeaders.CORRELATION_ID]: 'correlation-id',
+        },
+      };
+      expect(kafkaParser.parse(message)).to.deep.eq(expectedParsedMessage);
+      // Parse message again and verify it still works correctly
+      expect(kafkaParser.parse(message)).to.deep.eq(expectedParsedMessage);
+      // Verify message was not modified
+      expect(message).to.deep.eq({
+        headers: {
+          [KafkaHeaders.CORRELATION_ID]: Buffer.from('correlation-id'),
+        },
+        value: Buffer.from(JSON.stringify({ prop: 'value' })),
+        key: Buffer.from('1'),
+      });
+    });
   });
 });