Как реализовать задержку REST клиента до получения ответа из Kafka с учетом асинхронности

Что проверяют

Понимаете ли вы, что синхронный REST поверх асинхронной Kafka требует корреляции запроса и ответа, и что ждать можно, не занимая поток. Следом почти всегда спрашивают: «А если сервис запущен в трёх экземплярах, ответ придёт туда, где его ждут?»

Как отвечать

Схема: контроллер генерирует correlationId, кладёт его в заголовок сообщения вместе с топиком для ответа, отправляет запрос и возвращает незавершённый CompletableFuture. Потребитель ответов по тому же идентификатору находит future и завершает его.

• готовое решение: ReplyingKafkaTemplate из Spring Kafka. Метод sendAndReceive() сам проставляет KafkaHeaders.CORRELATION_ID и KafkaHeaders.REPLY_TOPIC и возвращает RequestReplyFuture. На стороне обработчика @KafkaListener с @SendTo, время ожидания задаёт setDefaultReplyTimeout(Duration);
• вручную: ConcurrentHashMap<String, CompletableFuture<Reply>>, orTimeout(5, TimeUnit.SECONDS) и удаление записи в whenComplete, иначе карта течёт;
• не блокировать поток: Spring MVC принимает CompletableFuture или DeferredResult и освобождает поток сервлета до ответа. На Java 21 с spring.threads.virtual.enabled=true (Spring Boot 3.2 и новее) блокирующее ожидание на виртуальных потоках тоже допустимо;
• таймаут для клиента: по истечении отдают 504 или 202 с идентификатором операции.

Несколько экземпляров: у каждого свой топик ответов, своя партиция (заголовок KafkaHeaders.REPLY_PARTITION) или уникальный group.id у контейнера ответов. Сильный ход: сказать, что часто лучше не держать соединение вовсе и отдавать 202 Accepted со ссылкой на статус, SSE или WebSocket.

Где спотыкаются

Ставят @Async на метод и вызывают его из того же класса. Вызов идёт мимо прокси, метод выполняется синхронно в текущем потоке.

Ждут через Thread.sleep в цикле или future.get() без таймаута. Потоки пула кончаются, и сервис перестаёт отвечать даже на простые запросы.

Забывают про масштабирование: ответ приходит в экземпляр, где future нет, запрос висит до таймаута, а в логах всё выглядит нормально.