Как синхронно ждать события из Kafka без разрыва соединения с клиентом

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

Умеете ли вы связать синхронный HTTP с асинхронной шиной: принять запрос, отправить команду в Kafka, дождаться ответного события и вернуть его клиенту, не держа поток и не теряя ответ. Следом спрашивают, что будет при нескольких инстансах сервиса и что делать, если ответ так и не пришёл.

Как отвечать

Основа это request-reply с correlation id. Каждой команде присваивается идентификатор, ответный сервис кладёт его в заголовок ответа, а у отправителя висит карта «идентификатор → CompletableFuture», которую завершает слушатель ответного топика. В Spring Kafka это готово в ReplyingKafkaTemplate: sendAndReceive возвращает RequestReplyFuture, заголовки KafkaHeaders.CORRELATION_ID и KafkaHeaders.REPLY_TOPIC проставляются сами, ожидание ограничено replyTimeout.

Чтобы соединение с клиентом не рвалось и потоки не кончались:

• асинхронный контроллер: вернуть CompletableFuture или DeferredResult из Spring MVC или Mono в WebFlux. Поток Tomcat освобождается, соединение ждёт ответа. Таймаут задаётся через spring.mvc.async.request-timeout;
• согласованные таймауты: ожидание ответа короче таймаута балансировщика и прокси, у nginx proxy_read_timeout по умолчанию 60 секунд. Иначе клиент получит 504, а результат уйдёт в пустоту;
• долгая обработка: если ответ может идти минутами, синхронное ожидание не подходит. Отдайте 202 Accepted с адресом статуса или шлите результат через SSE (SseEmitter) либо WebSocket.

При нескольких инстансах ответ должен прийти туда, где висит соединение. Варианты: у каждого инстанса своя группа консьюмеров на ответный топик и чужие correlation id отбрасываются, либо своя партиция, переданная в KafkaHeaders.REPLY_PARTITION.

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

Создают KafkaConsumer на каждый HTTP-запрос. Вход в группу занимает секунды, а при общем group.id ещё и ребалансирует остальных.

Блокируют поток через future.get() без таймаута. При 200 потоках Tomcat по умолчанию несколько сотен ждущих запросов кладут сервис целиком.

Не чистят карту ожиданий. Запрос, ответ на который не пришёл, оставляет запись в памяти навсегда, и карта растёт до OutOfMemoryError.