Início / user-11878640

Rayyan's questions

Martin Hope
Rayyan
Asked: 2024-12-14 02:41:32 +0800 CST
  • 5

Tenha um Spring Integrationaplicativo que consiste em múltiplos integration flowse, juntos, o aplicativo é implantado como um contêiner em Kubernetes. O contêiner é dimensionado automaticamente conforme a carga.

Entre esses vários fluxos de integração, um desses fluxos tenta novamente os registros com falha do BD e, a qualquer momento, apenas uma instância dele deve ser executada (como a Quartzintegração em Spring Batch).

Como esse comportamento pode ser implementado em Spring Integration?

spring-integration
Martin Hope
Rayyan
Asked: 2024-11-01 16:50:34 +0800 CST
  • 6

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
Martin Hope
Rayyan
Asked: 2024-06-24 23:29:35 +0800 CST
  • 6

Este problema é uma extensão do problema SO Spring Integration Restore ReactorContext into Imperative Processing

Qualquer exceção lançada após a chamada do WebFlux via WebFluxRequestExecutingMessageHandlerestá faltando a Traceinformação.

Amostra que descreve o problema disponível em https://github.com/syedyusufh/simple-handle-reactive.git

Isto é o que está acontecendo.

  1. Chamada WebFlux concluída, mas há erros de negócios na resposta.
  2. O Transformer tenta converter a resposta em Object e, em seguida, procura por erros de negócios; nesse caso, lança o usuário definido para o mesmo.
  3. Esta exceção lançada primeiro é registrada automaticamente por meio de WebFluxRequestExecutingMessageHandleronde o contexto é perdido
  4. Em seguida, a exceção é encaminhada para o errorChannel definido pelo aplicativo, que cria um novo traceId

Exceção:

2m2024-06-24T20:27:02.445+04:00  INFO 1436 --- [   scheduling-1] [66799e5608c967d694f450b82077c38c-fadc3f15d16b6b11] c.i.sample.config.TracingConfig          : Request Headers: [Content-Type:"application/json", traceparent:"00-66799e5608c967d694f450b82077c38c-fadc3f15d16b6b11-00"]
2024-06-24T20:27:03.997+04:00  INFO 1436 --- [ctor-http-nio-3] [66799e5608c967d694f450b82077c38c-fadc3f15d16b6b11] c.i.sample.config.TracingConfig          : Response Headers: [Access-Control-Allow-Origin:"*", Alt-Svc:"h3=":443"; ma=2592000", Content-Type:"application/json", Date:"Mon, 24 Jun 2024 16:27:05 GMT", Server:"Caddy", Vary:"Accept-Encoding", Transfer-Encoding:"chunked"]
2024-06-24T20:27:04.089+04:00 ERROR 1436 --- [oundedElastic-1] [                                                 ] .o.WebFluxRequestExecutingMessageHandler : Failed to send async reply: org.springframework.integration.support.MessageBuilder@5e1691b0
spring-integration