rest - 如何在 kafka 流应用程序中进行 REST API 调用/

Question

时间序列数据是在一个 kafka 主题中生成的。我需要读取每条记录并使用数据库中的一些数据进行装饰，并最终调用 REST API。收到响应后，输出到 kafka 主题。如何使用 kafka 流 API 高效且可扩展地做到这一点？

脚步 -

• 开始阅读输入主题
• 调用 mapvalues 进行数据库调用并用附加数据装饰记录
• 使用输入请求进行 REST api 调用，获取响应。
• 输出kafka topic中的记录

我认为，上述算法有两个瓶颈 -

• 进行数据库调用会减慢速度。这可以通过缓存元数据并在存在错误或使用状态存储时加载元数据来规避。

• 同步调用 REST API 会减慢速度。

    final KStream<String, String> records = builder.stream(InputTopic);

    //This is bad
    final KStream<String, String> output = records
      .mapValues(value ->  { //cache hit otherwise database call});
      .mapValues(value ->  { //prepare http request and convert the http resonse };
    output.to(OutputTopic)

如果数据库调用或 REST API 需要更长的时间来完成，上面的代码将对吞吐量产生依赖性和不利影响。具有相同键的记录不应乱序处理。预计吞吐量约为 1m/分钟。当一条记录到达 REST API 时，可以同时进行数据库调用。

不确定如何编写可以在这种情况下扩展的拓扑。我是卡夫卡流的新手。