Как совершать вызовы REST API в приложении kafka streams/

Данные временных рядов создаются в теме кафки. Мне нужно прочитать каждую запись и украсить некоторыми данными из базы данных и в конечном итоге вызвать REST API. Как только ответ получен, вывод в тему кафки. Как я могу сделать это с помощью API потоков Kafka эффективно и масштабируемо?

Шаги -

• Начать чтение входной темы
• Вызов значений карт, чтобы сделать вызов базы данных и украсить запись дополнительными данными.
• Сделайте вызов API REST с запросом ввода, получите ответ.
• Вывести запись в тему кафки

Я думаю, что в приведенном выше алгоритме есть два узких места -

• Выполнение вызовов базы данных замедлит его. Это можно обойти, кэшируя метаданные и загружая метаданные, когда происходит ошибка или используется хранилище состояний.

• Синхронный вызов 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 займет больше времени. Записи с одинаковым ключом не должны обрабатываться не по порядку. Ожидаемая пропускная способность составляет около 1 м/мин. Когда одна запись достигает REST API, можно одновременно выполнять вызовы базы данных.

Не знаю, как приступить к написанию топологии, которая может масштабироваться в этом сценарии. Я новичок в потоках кафки.