Início / coding / Perguntas / 79147403
Accepted
Rayyan
Asked: 2024-11-01 16:50:34 +0800 CST

Integração de primavera Kafka tentativa atrasada

  • 772

O caso de uso é reprocessar as mensagens com falha downstream com um atraso usando o Kafka do Spring Integration ConcurrentMessageListenerContainer. Digamos que o máximo de tentativas de repetição deve ser 2 com um atraso fixo de 5 minutos.

Existe uma solução pronta para uso disponível no framework Spring Integration?

Já revisei DeadLetterPublishingRecoverer, isso ajuda apenas na mudança para um DLT, sem processamento adicional, a menos que ouçamos o mesmo DLT com um poller.

spring-integration

1 respostas

  1. Best Answer

    Se você analisar uma solução de integração Spring, então KafkaMessageDrivenChannelAdapterqual usa o mencionado ConcurrentMessageListenerContainerpoderia ser configurado com RetryTemplate:

    /**
 * Specify a {@link RetryTemplate} instance to use for retrying deliveries.
 * <p>
 * IMPORTANT: This form of retry is blocking and could cause a rebalance if the
 * aggregate retry delays across all polled records might exceed the
 * {@code max.poll.interval.ms}. Instead, consider adding a
 * {@code DefaultErrorHandler} to the listener container, configured with a
 * {@link KafkaErrorSendingMessageRecoverer}.
 * @param retryTemplate the {@link RetryTemplate} to use.
 */
public void setRetryTemplate(RetryTemplate retryTemplate) {

    Outra maneira com Spring Integration é tentar novamente exatamente no lugar onde você manipula a mensagem downstream. Para isso, um endpoint específico pode ser configurado com um RequestHandlerRetryAdvice: https://docs.spring.io/spring-integration/reference/handler-advice/classes.html#retry-advice

    ATUALIZAR

    Existe um equivalente @EnableKafkaRetryTopicem Spring Integration.

    Não, não há como aplicar o @RetryableTopicno Spring Integration's KafkaMessageDrivenChannelAdapter. O problema é que essa nova tentativa não bloqueante é projetada exatamente para POJOs declarativos baseados em anotações. Onde o adaptador de canal é otimizado para MessageListenerinjeção regular no contêiner do listener de destino.

    Eu sugiro olhar para um @KafkaListenercom isso @RetryableTopicem vez disso e usar @MessagingGatewaypara chamar o fluxo de integração do Spring a partir desse método ouvinte: https://docs.spring.io/spring-integration/reference/gateway.html . Você pode até mesmo esperar um Message<?>em seu @KafkaListenere enviá-lo para um MessageChannel.

    • 2

relate perguntas