kafka上的NestJs cqrs异常和错误处理答案

【问题标题】：NestJs cqrs exception and error handling on kafkakafka上的NestJs cqrs异常和错误处理
【发布时间】：2020-06-29 17:25:31
【问题描述】：

我有点问题，

我在 NestJS 中使用带有 CQRS 的 Kafka。

我的问题是：在写入过程中处理错误的最佳方法是什么？

我通过 web api 向我的 nestjs 应用程序发送请求
我有并且事件ObjectedCreatedEvent在创建后将它发送到事件总线上并将其写入kafka（confluent.cloud）。
Kafka 响应错误 87，因为消息验证针对一个 json 架构失败（到目前为止还可以）
如何正确响应发生错误的 web api？

刚刚在此过程中启动了另一个侦听器？我希望 kafka（融合云）至少必须针对此类事情进行讨论。

我只要有一个 Event ObjectValdiationFailedEvent 并将它放到 eventbus 上。

【问题讨论】：

标签： apache-kafka nestjs cqrs confluent-cloud

【解决方案1】：

Confluent Cloud 在这里对您没有多大帮助，因为这是一个客户端问题。如您所知； Kafka 接收生产者序列化的任何内容，并将数据简单地存储到选定的分区中。在您的情况下，数据不是离开客户端的事件，这意味着抛出此 错误 87 的东西肯定不是服务器端 Kafka，而是客户端。

我的建议是设置您的框架 NestJS 支持的任何异常处理程序。我不是 Node.js 开发人员（我的背景是 Java 和 Go），但快速查看 NestJS 文档表明该框架允许您注册能够处理异常的过滤器。例如：

import { Catch, RpcExceptionFilter, ArgumentsHost } from '@nestjs/common';
import { Observable, throwError } from 'rxjs';
import { RpcException } from '@nestjs/microservices';

@Catch(RpcException)
export class ExceptionFilter implements RpcExceptionFilter<RpcException> {
  catch(exception: RpcException, host: ArgumentsHost): Observable<any> {
    return throwError(exception.getError());
  }
}

更多信息here。

因此，您可能需要调查哪个层抛出此错误 87，以便进行相应处理。

【讨论】：

错误 87 来自 kafka，它是对消息的模式验证，它是新定义的一部分 kafka.apache.org/protocol.html INVALID_RECORD | 87 |该记录未通过代理验证，因此被拒绝。
这很有趣。这意味着生成的记录甚至与 Kafka API 不兼容，因此这不是模式验证问题。
这是一个非常普遍的错误，但由于 avro 模式验证而发生在我身上。客户端在nestjs 或kafkajs 中没有正确处理它。无论如何，我的问题更多的是如何在 cqrs 中正确获得响应，而我的错是我认为它必须是一个同步响应，而 cqrs 中没有。所以我要做的就是自己在kafka上发送一个事件，让消费者监听它，前端将不得不异步接收它，或者通过等待kafka总线上的请求响应周期使异步事件同步的微服务。跨度>