Expose KafkaJS Event Listeners

[jgoldsmith613]

Is there an existing issue that is already proposing this?

• I have searched the existing issues

Is your feature request related to a problem? Please describe it

There is an error within kafkaJS that when the client has a generation ID error the consumer does not restart or consume anymore messages.

Here is a github issue describing the issue: tulios/kafkajs#1466
And here is a potential PR to fix it: tulios/kafkajs#1474

One of the mitigations to deal with the error until fixed require something like this:
kafkaClient.consumer.on("consumer.crash", (event) => {
if (event.payload.error.name === "KafkaJSNonRetriableError") {

     process.exit(1);  // will initiate a k8s restart

    // ... or do something else like reconnecting and starting run again ... 
}

});

But I don't believe Nest exposes access to the kafka client like that to be able to do this.

Describe the solution you'd like

Add the ability register event listeners on the kafka client.

the cause of this issue KafkaJS, but even if they fix this current issue where there client, this could still be a useful feature.

Teachability, documentation, adoption, migration strategy

Modify nestjs kafka docs to include example of how to add listener

What is the motivation / use case for changing the behavior?

Initially its because of KafkaJS issue, but could be useful elsewhere too.

[kamilmysliwiec]

Would you like to create a PR for this issue?

[jgoldsmith613]

Would you like to create a PR for this issue?

Sorry for the delay here, we figured out a way to get access to the kafka client by doing the following.

app
    .getMicroservices()
    .filter((service: any) => service.server.transportId === Transport.KAFKA)
    .forEach((service: any) => {
      const consumer: kafkajs.Consumer = service.server.consumer

      consumer.on(consumer.events.CRASH, event => {