feat: [CHK-3347] Added logic to warmup lazy beans (#365)

* added logic to warmup lazy beans * spotless apply * done * fixed * removed unused refs * spotless apply * chore: add warmup logging * updated * added more warmup stubs * reused dummy checkpointer class * all jsons ready * warmup functions ready * spotless apply * junit test need more work * updated unit tests * updated unit tests * spotless * Update WarmupRequests.kt * Update WarmupRequestsTest.kt * Update ServicesWarmup.kt * Delete src/test/kotlin/it/pagopa/ecommerce/eventdispatcher/warmup/ServicesWarmupTest.kt * fix * Update StreamConfig.kt * refactored used object * Delete src/test/kotlin/it/pagopa/ecommerce/eventdispatcher/warmup/ServicesRequestsTest.kt * Update WarmupRequests.kt * Create EventUtils * Update WarmupRequests.kt * Create EventsUtil * Delete src/test/kotlin/it/pagopa/ecommerce/eventdispatcher/warmup/utils/EventUtils * fix * Update EventsUtil.java * fix * Update src/main/kotlin/it/pagopa/ecommerce/eventdispatcher/warmup/ServicesWarmup.kt Co-authored-by: Pietro Tota <[email protected]> * fix * fix: add tests mongo connection string parameter for connection timeout * fix: add programmatic redis stream message receiver startup --------- Co-authored-by: Pietro Tota <[email protected]> Co-authored-by: Pietro Tota <[email protected]>
pagopa · Jan 17, 2025 · c4d6d68 · c4d6d68
1 parent 9654a5a
commit c4d6d68
Show file tree

Hide file tree

Showing 19 changed files with 687 additions and 14 deletions.
diff --git a/src/main/kotlin/it/pagopa/ecommerce/eventdispatcher/config/QueuesConsumerConfig.kt b/src/main/kotlin/it/pagopa/ecommerce/eventdispatcher/config/QueuesConsumerConfig.kt
@@ -21,6 +21,7 @@ class QueuesConsumerConfig {
   @Bean
   @InboundChannelAdapter(
     channel = "transactionclosureschannel",
+    autoStartup = "false",
     poller = [Poller(fixedDelay = "1000", maxMessagesPerPoll = "10")])
   @EndpointId("storageQueueClosuresMessageSourceEndpoint")
   fun storageQueueClosuresMessageSource(
@@ -33,6 +34,7 @@ class QueuesConsumerConfig {
   @Bean
   @InboundChannelAdapter(
     channel = "transactionretryclosureschannel",
+    autoStartup = "false",
     poller = [Poller(fixedDelay = "1000", maxMessagesPerPoll = "10")])
   @EndpointId("storageQueueRetryClosuresMessageSourceEndpoint")
   fun storageQueueRetryClosuresMessageSource(
@@ -46,6 +48,7 @@ class QueuesConsumerConfig {
   @Bean
   @InboundChannelAdapter(
     channel = "transactionexpiredchannel",
+    autoStartup = "false",
     poller = [Poller(fixedDelay = "1000", maxMessagesPerPoll = "10")])
   @EndpointId("storageQueueExpirationsMessageSourceEndpoint")
   fun storageQueueExpirationsMessageSource(
@@ -58,6 +61,7 @@ class QueuesConsumerConfig {
   @Bean
   @InboundChannelAdapter(
     channel = "transactionrefundretrychannel",
+    autoStartup = "false",
     poller = [Poller(fixedDelay = "1000", maxMessagesPerPoll = "10")])
   @EndpointId("storageQueueRefundRetryMessageSourceEndpoint")
   fun storageQueueRefundRetryMessageSource(
@@ -70,6 +74,7 @@ class QueuesConsumerConfig {
   @Bean
   @InboundChannelAdapter(
     channel = "transactionsrefundchannel",
+    autoStartup = "false",
     poller = [Poller(fixedDelay = "1000", maxMessagesPerPoll = "10")])
   @EndpointId("storageQueueRefundMessageSourceEndpoint")
   fun storageQueueRefundMessageSource(
@@ -82,6 +87,7 @@ class QueuesConsumerConfig {
   @Bean
   @InboundChannelAdapter(
     channel = "transactionretrynotificationschannel",
+    autoStartup = "false",
     poller = [Poller(fixedDelay = "1000", maxMessagesPerPoll = "10")])
   @EndpointId("storageQueueRetryNotificationsMessageSourceEndpoint")
   fun storageQueueRetryNotificationsMessageSource(
@@ -96,6 +102,7 @@ class QueuesConsumerConfig {
   @Bean
   @InboundChannelAdapter(
     channel = "transactionnotificationschannel",
+    autoStartup = "false",
     poller = [Poller(fixedDelay = "1000", maxMessagesPerPoll = "10")])
   @EndpointId("storageQueueNotificationsMessageSourceEndpoint")
   fun storageQueueNotificationsMessageSource(
@@ -108,6 +115,7 @@ class QueuesConsumerConfig {
   @Bean
   @InboundChannelAdapter(
     channel = "transactionsauthorizationrequestedchannel",
+    autoStartup = "false",
     poller = [Poller(fixedDelay = "1000", maxMessagesPerPoll = "10")])
   @EndpointId("storageQueueAuthorizationRequestedMessageSourceEndpoint")
   fun storageQueueAuthorizationRequestedMessageSource(
@@ -121,6 +129,7 @@ class QueuesConsumerConfig {
   @Bean
   @InboundChannelAdapter(
     channel = "transactionsauthorizationoutcomewaitingchannel",
+    autoStartup = "false",
     poller = [Poller(fixedDelay = "1000", maxMessagesPerPoll = "10")])
   @EndpointId("storageQueueAuthorizationOutcomeWaitingMessageSourceEndpoint")
   fun storageQueueAuthorizationOutcomeWaitingMessageSource(

diff --git a/src/main/kotlin/it/pagopa/ecommerce/eventdispatcher/config/redis/stream/StreamConfig.kt b/src/main/kotlin/it/pagopa/ecommerce/eventdispatcher/config/redis/stream/StreamConfig.kt
@@ -14,6 +14,7 @@ import org.springframework.data.redis.hash.Jackson2HashMapper
 import org.springframework.data.redis.serializer.RedisSerializationContext
 import org.springframework.data.redis.serializer.StringRedisSerializer
 import org.springframework.data.redis.stream.StreamReceiver
+import org.springframework.integration.annotation.EndpointId
 import org.springframework.integration.annotation.InboundChannelAdapter
 import org.springframework.integration.annotation.Poller
 
@@ -34,7 +35,7 @@ class StreamConfig {
           EventDispatcherGenericCommand::class.java, EventDispatcherCommandMixin::class.java)
     val streamReceiverOptions =
       StreamReceiver.StreamReceiverOptions.builder()
-        .pollTimeout(Duration.ofMillis(100))
+        .pollTimeout(Duration.ofMillis(1000))
         .batchSize(1) // read one item per poll
         .keySerializer(
           RedisSerializationContext.SerializationPair.fromSerializer(StringRedisSerializer()))
@@ -51,7 +52,9 @@ class StreamConfig {
   @Bean
   @InboundChannelAdapter(
     channel = "eventDispatcherReceiverCommandChannel",
-    poller = [Poller(fixedDelay = "1000", maxMessagesPerPoll = "1")])
+    poller = [Poller(fixedDelay = "1000", maxMessagesPerPoll = "1")],
+    autoStartup = "false")
+  @EndpointId("eventDispatcherReceiverCommandChannelEndpoint")
   fun eventDispatcherReceiverCommandMessageSource(
     redisStreamReceiver: StreamReceiver<String, ObjectRecord<String, LinkedHashMap<*, *>>>,
     eventDispatcherCommandsTemplateWrapper: EventDispatcherCommandsTemplateWrapper,

diff --git a/...a/ecommerce/eventdispatcher/queues/TransactionAuthorizationOutcomeWaitingQueueConsumer.kt b/...a/ecommerce/eventdispatcher/queues/TransactionAuthorizationOutcomeWaitingQueueConsumer.kt
@@ -9,6 +9,9 @@ import it.pagopa.ecommerce.commons.queues.QueueEvent
 import it.pagopa.ecommerce.commons.queues.StrictJsonSerializerProvider
 import it.pagopa.ecommerce.eventdispatcher.exceptions.InvalidEventException
 import it.pagopa.ecommerce.eventdispatcher.utils.DeadLetterTracedQueueAsyncClient
+import it.pagopa.ecommerce.eventdispatcher.warmup.annotations.WarmupFunction
+import it.pagopa.ecommerce.payment.requests.warmup.utils.DummyCheckpointer
+import it.pagopa.ecommerce.payment.requests.warmup.utils.WarmupRequests.getTransactionAuthorizationOutcomeWaitingEvent
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
@@ -85,4 +88,9 @@ class TransactionAuthorizationOutcomeWaitingQueueConsumer(
           DeadLetterTracedQueueAsyncClient.PARSING_EVENT_ERROR_CONTEXT)
       }
   }
+
+  @WarmupFunction
+  fun warmupService() {
+    messageReceiver(getTransactionAuthorizationOutcomeWaitingEvent(), DummyCheckpointer).block()
+  }
 }
diff --git a/...pagopa/ecommerce/eventdispatcher/queues/TransactionAuthorizationRequestedQueueConsumer.kt b/...pagopa/ecommerce/eventdispatcher/queues/TransactionAuthorizationRequestedQueueConsumer.kt
@@ -9,6 +9,9 @@ import it.pagopa.ecommerce.commons.queues.QueueEvent
 import it.pagopa.ecommerce.commons.queues.StrictJsonSerializerProvider
 import it.pagopa.ecommerce.eventdispatcher.exceptions.InvalidEventException
 import it.pagopa.ecommerce.eventdispatcher.utils.DeadLetterTracedQueueAsyncClient
+import it.pagopa.ecommerce.eventdispatcher.warmup.annotations.WarmupFunction
+import it.pagopa.ecommerce.payment.requests.warmup.utils.DummyCheckpointer
+import it.pagopa.ecommerce.payment.requests.warmup.utils.WarmupRequests.getTransactionAuthorizationRequestedEvent
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
@@ -84,4 +87,9 @@ class TransactionAuthorizationRequestedQueueConsumer(
           DeadLetterTracedQueueAsyncClient.PARSING_EVENT_ERROR_CONTEXT)
       }
   }
+
+  @WarmupFunction
+  fun warmupService() {
+    messageReceiver(getTransactionAuthorizationRequestedEvent(), DummyCheckpointer).block()
+  }
 }
diff --git a/...kotlin/it/pagopa/ecommerce/eventdispatcher/queues/TransactionClosePaymentQueueConsumer.kt b/...kotlin/it/pagopa/ecommerce/eventdispatcher/queues/TransactionClosePaymentQueueConsumer.kt
@@ -14,6 +14,9 @@ import it.pagopa.ecommerce.commons.queues.StrictJsonSerializerProvider
 import it.pagopa.ecommerce.commons.queues.TracingInfo
 import it.pagopa.ecommerce.eventdispatcher.exceptions.*
 import it.pagopa.ecommerce.eventdispatcher.utils.DeadLetterTracedQueueAsyncClient
+import it.pagopa.ecommerce.eventdispatcher.warmup.annotations.WarmupFunction
+import it.pagopa.ecommerce.payment.requests.warmup.utils.DummyCheckpointer
+import it.pagopa.ecommerce.payment.requests.warmup.utils.WarmupRequests.getTransactionClosureRequestedEvent
 import java.util.*
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
@@ -126,4 +129,9 @@ class TransactionClosePaymentQueueConsumer(
           DeadLetterTracedQueueAsyncClient.PARSING_EVENT_ERROR_CONTEXT)
       }
   }
+
+  @WarmupFunction
+  fun warmupService() {
+    messageReceiver(getTransactionClosureRequestedEvent(), DummyCheckpointer).block()
+  }
 }
diff --git a/...n/it/pagopa/ecommerce/eventdispatcher/queues/TransactionClosePaymentRetryQueueConsumer.kt b/...n/it/pagopa/ecommerce/eventdispatcher/queues/TransactionClosePaymentRetryQueueConsumer.kt
@@ -16,6 +16,9 @@ import it.pagopa.ecommerce.commons.queues.StrictJsonSerializerProvider
 import it.pagopa.ecommerce.commons.queues.TracingInfo
 import it.pagopa.ecommerce.eventdispatcher.exceptions.InvalidEventException
 import it.pagopa.ecommerce.eventdispatcher.utils.DeadLetterTracedQueueAsyncClient
+import it.pagopa.ecommerce.eventdispatcher.warmup.annotations.WarmupFunction
+import it.pagopa.ecommerce.payment.requests.warmup.utils.DummyCheckpointer
+import it.pagopa.ecommerce.payment.requests.warmup.utils.WarmupRequests.getTransactionClosureErrorEvent
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
@@ -163,4 +166,10 @@ class TransactionClosePaymentRetryQueueConsumer(
           DeadLetterTracedQueueAsyncClient.PARSING_EVENT_ERROR_CONTEXT)
       }
   }
+
+  @WarmupFunction
+  fun warmupService() {
+    messageReceiver(getTransactionClosureErrorEvent(), DummyCheckpointer, EmptyTransaction())
+      .block()
+  }
 }
diff --git a/...n/kotlin/it/pagopa/ecommerce/eventdispatcher/queues/TransactionExpirationQueueConsumer.kt b/...n/kotlin/it/pagopa/ecommerce/eventdispatcher/queues/TransactionExpirationQueueConsumer.kt
@@ -17,6 +17,9 @@ import it.pagopa.ecommerce.eventdispatcher.exceptions.InvalidEventException
 import it.pagopa.ecommerce.eventdispatcher.queues.v1.TransactionExpirationQueueConsumer as TransactionExpirationQueueConsumerV1
 import it.pagopa.ecommerce.eventdispatcher.queues.v2.TransactionExpirationQueueConsumer as TransactionExpirationQueueConsumerV2
 import it.pagopa.ecommerce.eventdispatcher.utils.DeadLetterTracedQueueAsyncClient
+import it.pagopa.ecommerce.eventdispatcher.warmup.annotations.WarmupFunction
+import it.pagopa.ecommerce.payment.requests.warmup.utils.DummyCheckpointer
+import it.pagopa.ecommerce.payment.requests.warmup.utils.WarmupRequests.getTransactionExpiredEvent
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
@@ -171,4 +174,10 @@ class TransactionExpirationQueueConsumer(
           DeadLetterTracedQueueAsyncClient.PARSING_EVENT_ERROR_CONTEXT)
       }
   }
+
+  @WarmupFunction
+  fun warmupService() {
+    messageReceiver(getTransactionExpiredEvent(), DummyCheckpointer, MessageHeaders(emptyMap()))
+      .block()
+  }
 }
diff --git a/...otlin/it/pagopa/ecommerce/eventdispatcher/queues/TransactionNotificationsQueueConsumer.kt b/...otlin/it/pagopa/ecommerce/eventdispatcher/queues/TransactionNotificationsQueueConsumer.kt
@@ -14,6 +14,9 @@ import it.pagopa.ecommerce.commons.queues.StrictJsonSerializerProvider
 import it.pagopa.ecommerce.commons.queues.TracingInfo
 import it.pagopa.ecommerce.eventdispatcher.exceptions.InvalidEventException
 import it.pagopa.ecommerce.eventdispatcher.utils.DeadLetterTracedQueueAsyncClient
+import it.pagopa.ecommerce.eventdispatcher.warmup.annotations.WarmupFunction
+import it.pagopa.ecommerce.payment.requests.warmup.utils.DummyCheckpointer
+import it.pagopa.ecommerce.payment.requests.warmup.utils.WarmupRequests.getTransactionUserReceiptRequestedEvent
 import it.pagopa.generated.notifications.templates.success.*
 import java.util.*
 import org.slf4j.Logger
@@ -115,4 +118,9 @@ class TransactionNotificationsQueueConsumer(
           DeadLetterTracedQueueAsyncClient.PARSING_EVENT_ERROR_CONTEXT)
       }
   }
+
+  @WarmupFunction
+  fun warmupService() {
+    messageReceiver(getTransactionUserReceiptRequestedEvent(), DummyCheckpointer).block()
+  }
 }
diff --git a/.../it/pagopa/ecommerce/eventdispatcher/queues/TransactionNotificationsRetryQueueConsumer.kt b/.../it/pagopa/ecommerce/eventdispatcher/queues/TransactionNotificationsRetryQueueConsumer.kt
@@ -14,10 +14,10 @@ import it.pagopa.ecommerce.commons.queues.QueueEvent
 import it.pagopa.ecommerce.commons.queues.StrictJsonSerializerProvider
 import it.pagopa.ecommerce.commons.queues.TracingInfo
 import it.pagopa.ecommerce.eventdispatcher.exceptions.InvalidEventException
-import it.pagopa.ecommerce.eventdispatcher.queues.*
 import it.pagopa.ecommerce.eventdispatcher.utils.DeadLetterTracedQueueAsyncClient
-import it.pagopa.generated.notifications.templates.success.*
-import java.util.*
+import it.pagopa.ecommerce.eventdispatcher.warmup.annotations.WarmupFunction
+import it.pagopa.ecommerce.payment.requests.warmup.utils.DummyCheckpointer
+import it.pagopa.ecommerce.payment.requests.warmup.utils.WarmupRequests.getTransactionUserReceiptAddErrorEvent
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
@@ -164,4 +164,9 @@ class TransactionNotificationsRetryQueueConsumer(
           DeadLetterTracedQueueAsyncClient.PARSING_EVENT_ERROR_CONTEXT)
       }
   }
+
+  @WarmupFunction
+  fun warmupService() {
+    messageReceiver(getTransactionUserReceiptAddErrorEvent(), DummyCheckpointer).block()
+  }
 }
diff --git a/.../kotlin/it/pagopa/ecommerce/eventdispatcher/queues/TransactionRefundRetryQueueConsumer.kt b/.../kotlin/it/pagopa/ecommerce/eventdispatcher/queues/TransactionRefundRetryQueueConsumer.kt
@@ -12,6 +12,9 @@ import it.pagopa.ecommerce.commons.queues.StrictJsonSerializerProvider
 import it.pagopa.ecommerce.commons.queues.TracingInfo
 import it.pagopa.ecommerce.eventdispatcher.exceptions.InvalidEventException
 import it.pagopa.ecommerce.eventdispatcher.utils.DeadLetterTracedQueueAsyncClient
+import it.pagopa.ecommerce.eventdispatcher.warmup.annotations.WarmupFunction
+import it.pagopa.ecommerce.payment.requests.warmup.utils.DummyCheckpointer
+import it.pagopa.ecommerce.payment.requests.warmup.utils.WarmupRequests.getTransactionRefundRetriedEvent
 import java.util.*
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
@@ -119,4 +122,9 @@ class TransactionRefundRetryQueueConsumer(
           DeadLetterTracedQueueAsyncClient.PARSING_EVENT_ERROR_CONTEXT)
       }
   }
+
+  @WarmupFunction
+  fun warmupService() {
+    messageReceiver(getTransactionRefundRetriedEvent(), DummyCheckpointer).block()
+  }
 }
diff --git a/...main/kotlin/it/pagopa/ecommerce/eventdispatcher/queues/TransactionsRefundQueueConsumer.kt b/...main/kotlin/it/pagopa/ecommerce/eventdispatcher/queues/TransactionsRefundQueueConsumer.kt
@@ -15,6 +15,9 @@ import it.pagopa.ecommerce.commons.queues.StrictJsonSerializerProvider
 import it.pagopa.ecommerce.commons.queues.TracingInfo
 import it.pagopa.ecommerce.eventdispatcher.exceptions.InvalidEventException
 import it.pagopa.ecommerce.eventdispatcher.utils.DeadLetterTracedQueueAsyncClient
+import it.pagopa.ecommerce.eventdispatcher.warmup.annotations.WarmupFunction
+import it.pagopa.ecommerce.payment.requests.warmup.utils.DummyCheckpointer
+import it.pagopa.ecommerce.payment.requests.warmup.utils.WarmupRequests.getTransactionRefundRequestedEvent
 import java.util.*
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
@@ -167,4 +170,9 @@ class TransactionsRefundQueueConsumer(
           DeadLetterTracedQueueAsyncClient.PARSING_EVENT_ERROR_CONTEXT)
       }
   }
+
+  @WarmupFunction
+  fun warmupService() {
+    messageReceiver(getTransactionRefundRequestedEvent(), DummyCheckpointer).block()
+  }
 }
diff --git a/src/main/kotlin/it/pagopa/ecommerce/eventdispatcher/queues/v2/common.kt b/src/main/kotlin/it/pagopa/ecommerce/eventdispatcher/queues/v2/common.kt
@@ -976,6 +976,7 @@ fun <T> runTracedPipelineWithDeadLetterQueue(
   spanName: String,
   jsonSerializerProviderV2: StrictJsonSerializerProvider
 ): Mono<Unit> {
+  val nullTransactionId = "00000000000000000000000000000000" // null event ID used in warmup phase
   val eventLogString = "${queueEvent.event.id}, transactionId: ${queueEvent.event.transactionId}"
 
   val deadLetterPipeline =
@@ -995,14 +996,19 @@ fun <T> runTracedPipelineWithDeadLetterQueue(
             else -> DeadLetterTracedQueueAsyncClient.ErrorCategory.PROCESSING_ERROR
           }
         logger.error("Exception processing event $eventLogString", pipelineException)
-        deadLetterTracedQueueAsyncClient.sendAndTraceDeadLetterQueueEvent(
-          binaryData = BinaryData.fromObject(queueEvent, jsonSerializerProviderV2.createInstance()),
-          errorContext =
-            DeadLetterTracedQueueAsyncClient.ErrorContext(
-              transactionId = TransactionId(queueEvent.event.transactionId),
-              transactionEventCode = queueEvent.event.eventCode,
-              errorCategory = errorCategory),
-        )
+        if (queueEvent.event.transactionId != nullTransactionId) {
+          deadLetterTracedQueueAsyncClient.sendAndTraceDeadLetterQueueEvent(
+            binaryData =
+              BinaryData.fromObject(queueEvent, jsonSerializerProviderV2.createInstance()),
+            errorContext =
+              DeadLetterTracedQueueAsyncClient.ErrorContext(
+                transactionId = TransactionId(queueEvent.event.transactionId),
+                transactionEventCode = queueEvent.event.eventCode,
+                errorCategory = errorCategory))
+        } else {
+          logger.info("Skipping dead letter queue for warmup event with null transaction ID")
+          Mono.just(Unit)
+        }
       }
 
   return tracingUtils

diff --git a/...pagopa/ecommerce/eventdispatcher/services/InboundChannelAdapterLifecycleHandlerService.kt b/...pagopa/ecommerce/eventdispatcher/services/InboundChannelAdapterLifecycleHandlerService.kt
@@ -65,4 +65,11 @@ class InboundChannelAdapterLifecycleHandlerService(
       .getBeansWithAnnotation(InboundChannelAdapter::class.java)
       .filterNot { it.value is RedisStreamMessageSource }
       .keys
+
+  fun invokeCommandForRedisStreamMessageSource(command: String) {
+    val controllerBusMessage =
+      MessageBuilder.createMessage(
+        "@eventDispatcherReceiverCommandChannelEndpoint.$command()", MessageHeaders(mapOf()))
+    controlBusInput.send(controllerBusMessage)
+  }
 }