Repository: nestjs/nest (ts)

Issue:
Expose Producer in KafkaContext
### Is there an existing issue that is already proposing this?

- [X] I have searched the existing issues

### Is your feature request related to a problem? Please describe it

When implementing various error handling strategies using Kafka (retry topic, dead letter queue) using an Exception filter, I need some way of producing messages in there - one way is injecting the client, which IMHO violates the separation of concerns and requires to import the clients module where the exception filter is used.

I found it useful to put the producer to the context object, so I can retrieve it in the exception filter and put the message to a retry topic or to the DLQ based on the error severity.

For that to work, I have to create a custom transporter by extending the base `KafkaServer` and overwriting a great deal of the code - basically, I have to copy-and-paste the entirety of the `handleMessage` method only to change the context class creation.

### Describe the solution you'd like

I see two solutions here:
1) Just include the producer in the KafkaContext
2) Create a protected `createContext` method that can be overriden by child classes

### Teachability, documentation, adoption, migration strategy

If the producer is added as the last parameter to the `KafkaContextArgs`, then the change will not be breaking at all (but it wouldn't look pretty) and it would immediately be easier to create retry strategies based on producing messages.

### What is the motivation / use case for changing the behavior?

Most serious Kafka implementations will require some kind of retry and dead letter topics, so exposing the producer in the context would make it easier to adopt the practice instead of working around the framework limitations.

Pull request:
feat(microservices): Add producer reference to KafkaContext
## 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". -->
- [ ] Bugfix
- [x] 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 producer is not exposed in `KafkaContext`

Issue Number:  #10263 


## What is the new behavior?
Kafka producer is exposed in `KafkaContext` under `getProducer()`
Closes #10263 

## Does this PR introduce a breaking change?
- [ ] Yes
- [x] No

In order not to introduce a braking change, I added the producer as the last argument of `KafkaContextArgs`. It would look better if it were immediately after `consumer`.

## Other information
Btw, I noticed that all Kafka related integration tests are skipped for being "flaky", therefore I didn't add one. Is this something that can be worked on to make them reliable?

Patch:
diff --git a/packages/microservices/ctx-host/kafka.context.ts b/packages/microservices/ctx-host/kafka.context.ts
index 7d1f5e4a5..4ba1663c8 100644
--- a/packages/microservices/ctx-host/kafka.context.ts
+++ b/packages/microservices/ctx-host/kafka.context.ts
@@ -1,4 +1,4 @@
-import { Consumer, KafkaMessage } from '../external/kafka.interface';
+import { Consumer, KafkaMessage, Producer } from '../external/kafka.interface';
 import { BaseRpcContext } from './base-rpc.context';
 
 type KafkaContextArgs = [
@@ -7,6 +7,7 @@ type KafkaContextArgs = [
   topic: string,
   consumer: Consumer,
   heartbeat: () => Promise<void>,
+  producer: Producer,
 ];
 
 export class KafkaContext extends BaseRpcContext<KafkaContextArgs> {
@@ -48,4 +49,11 @@ export class KafkaContext extends BaseRpcContext<KafkaContextArgs> {
   getHeartbeat() {
     return this.args[4];
   }
+
+  /**
+   * Returns the Kafka producer reference,
+   */
+  getProducer() {
+    return this.args[5];
+  }
 }
diff --git a/packages/microservices/server/server-kafka.ts b/packages/microservices/server/server-kafka.ts
index e7bfee7ce..7ac52f9fa 100644
--- a/packages/microservices/server/server-kafka.ts
+++ b/packages/microservices/server/server-kafka.ts
@@ -168,6 +168,7 @@ export class ServerKafka extends Server implements CustomTransportStrategy {
       payload.topic,
       this.consumer,
       payload.heartbeat,
+      this.producer,
     ]);
     const handler = this.getHandlerByPattern(packet.pattern);
     // if the correlation id or reply topic is not set


Test patch:
diff --git a/packages/microservices/test/ctx-host/kafka.context.spec.ts b/packages/microservices/test/ctx-host/kafka.context.spec.ts
index 9be85a12b..11ba73456 100644
--- a/packages/microservices/test/ctx-host/kafka.context.spec.ts
+++ b/packages/microservices/test/ctx-host/kafka.context.spec.ts
@@ -1,6 +1,10 @@
 import { expect } from 'chai';
 import { KafkaContext } from '../../ctx-host';
-import { Consumer, KafkaMessage } from '../../external/kafka.interface';
+import {
+  Consumer,
+  KafkaMessage,
+  Producer,
+} from '../../external/kafka.interface';
 
 describe('KafkaContext', () => {
   const args = [
@@ -9,12 +13,20 @@ describe('KafkaContext', () => {
     undefined,
     { test: 'consumer' },
     () => {},
+    { test: 'producer' },
   ];
   let context: KafkaContext;
 
   beforeEach(() => {
     context = new KafkaContext(
-      args as [KafkaMessage, number, string, Consumer, () => Promise<void>],
+      args as [
+        KafkaMessage,
+        number,
+        string,
+        Consumer,
+        () => Promise<void>,
+        Producer,
+      ],
     );
   });
   describe('getTopic', () => {
@@ -42,4 +54,9 @@ describe('KafkaContext', () => {
       expect(context.getHeartbeat()).to.be.eql(args[4]);
     });
   });
+  describe('getProducer', () => {
+    it('should return producer instance', () => {
+      expect(context.getProducer()).to.deep.eq({ test: 'producer' });
+    });
+  });
 });