Блог

в моем приложении kafka streams мне нужно повторно пытаться обрабатывать сообщение всякий раз, когда в логике обработки возникает исключение определенного типа.

Вместо того, чтобы оборачивать свою логику в RetryTemplate (я использую springboot), я рассматриваю возможность просто записать сообщение обратно в тему ввода, я предполагаю, что это сообщение будет добавлено в конец журнала в соответствующем разделе и в конечном итоге будет повторно обработано.

Я понимаю, что это испортило бы заказ, и меня это устраивает.

Мой вопрос в том, возникнет ли проблема у потоков кафки, когда он столкнется с сообщением, которое предположительно уже было обработано в прошлом (я предполагаю, что потоки кафки могут помечать обработанные сообщения, особенно когда это включено)?

Вот пример кода, который я рассматриваю для этого решения.

 val branches = streamsBuilder.stream(inputTopicName)
            .mapValues { it -> myServiceObject.executeSomeLogic(it) }
            .branch(
                { _, value -> value is successfulResult() }, // success
                { _, error -> error is errorResult() },  // exception was thrown
            )

branches[0].to(outputTopicName)
branches[1].to(inputTopicName) //write them back to input as a way of retrying