Pivotal ID # 186629008: Slack Message For Failed Notifications (#794)

https://www.pivotaltracker.com/story/show/186629008

- Add message converter to the rabbit template
- Configure custom exception handler to catch failing messages

scheduler/tasks/submission-releaser-task/src/main/kotlin/uk/ac/ebi/scheduler/releaser/config/ApplicationConfig.kt

@@ -5,6 +5,7 @@ import ac.uk.ebi.biostd.client.integration.web.SecurityWebClient
 import ac.uk.ebi.biostd.persistence.doc.MongoDbReactiveConfig
 import ac.uk.ebi.biostd.persistence.doc.db.reactive.repositories.SubmissionReleaserRepository
 import org.springframework.amqp.rabbit.core.RabbitTemplate
+import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
 import org.springframework.context.annotation.Import
@@ -16,8 +17,8 @@ import uk.ac.ebi.scheduler.releaser.service.SubmissionReleaserService
 @Configuration
 @Import(MongoDbReactiveConfig::class)
 class ApplicationConfig(
-    private val releaserRepository: SubmissionReleaserRepository,
     private val appProperties: ApplicationProperties,
+    private val releaserRepository: SubmissionReleaserRepository,
 ) {
     @Bean
     fun submissionReleaserService(
@@ -52,5 +53,8 @@ class ApplicationConfig(
     fun eventsPublisherService(
         rabbitTemplate: RabbitTemplate,
         eventsProperties: EventsProperties,
-    ): EventsPublisherService = EventsPublisherService(rabbitTemplate, eventsProperties)
+    ): EventsPublisherService {
+        rabbitTemplate.messageConverter = Jackson2JsonMessageConverter()
+        return EventsPublisherService(rabbitTemplate, eventsProperties)
+    }
 }

submission/submission-core/src/main/kotlin/ac/uk/ebi/biostd/submission/domain/submitter/ExtSubmissionSubmitter.kt

@@ -5,13 +5,22 @@ import ebi.ac.uk.extended.model.ExtSubmission
 
 interface ExtSubmissionSubmitter {
     suspend fun createRequest(rqt: ExtSubmitRequest): Pair<String, Int>
+
     suspend fun indexRequest(accNo: String, version: Int)
+
     suspend fun loadRequest(accNo: String, version: Int)
+
     suspend fun cleanRequest(accNo: String, version: Int)
+
     suspend fun processRequest(accNo: String, version: Int)
+
     suspend fun checkReleased(accNo: String, version: Int)
+
     suspend fun saveRequest(accNo: String, version: Int): ExtSubmission
+
     suspend fun finalizeRequest(accNo: String, version: Int): ExtSubmission
+
     suspend fun release(accNo: String)
+
     suspend fun handleRequest(accNo: String, version: Int): ExtSubmission
 }

submission/submission-handlers/src/main/kotlin/ac/uk/ebi/biostd/handlers/config/ApplicationConfig.kt

@@ -1,6 +1,7 @@
 package ac.uk.ebi.biostd.handlers.config
 
 import ac.uk.ebi.biostd.handlers.api.BioStudiesWebConsumer
+import ac.uk.ebi.biostd.handlers.exception.CustomErrorHandler
 import ac.uk.ebi.biostd.handlers.listeners.LogSubmissionListener
 import ac.uk.ebi.biostd.handlers.listeners.SecurityNotificationListener
 import ac.uk.ebi.biostd.handlers.listeners.SubmissionNotificationsListener
@@ -10,7 +11,12 @@ import ebi.ac.uk.commons.http.slack.NotificationsSender
 import ebi.ac.uk.notifications.integration.NotificationConfig
 import ebi.ac.uk.notifications.service.RtNotificationService
 import ebi.ac.uk.notifications.service.SecurityNotificationService
+import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory
+import org.springframework.amqp.rabbit.connection.ConnectionFactory
 import org.springframework.amqp.rabbit.core.RabbitTemplate
+import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler
+import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter
+import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
 import org.springframework.context.annotation.Import
@@ -25,6 +31,23 @@ class Listeners {
     @Bean
     fun logListener(): LogSubmissionListener = LogSubmissionListener()
 
+    @Bean
+    fun customErrorHandler(
+        notificationsSender: NotificationsSender,
+    ): ConditionalRejectingErrorHandler = CustomErrorHandler(notificationsSender)
+
+    @Bean
+    fun rabbitTemplate(
+        connectionFactory: ConnectionFactory,
+        factory: SimpleRabbitListenerContainerFactory,
+        errorHandler: ConditionalRejectingErrorHandler,
+        configurer: SimpleRabbitListenerContainerFactoryConfigurer,
+    ): RabbitTemplate {
+        factory.setErrorHandler(errorHandler)
+        configurer.configure(factory, connectionFactory)
+        return RabbitTemplate(connectionFactory).apply { messageConverter = Jackson2JsonMessageConverter() }
+    }
+
     @Bean
     fun notificationsListener(
         rabbitTemplate: RabbitTemplate,

submission/submission-handlers/src/main/kotlin/ac/uk/ebi/biostd/handlers/exception/CustomErrorHandler.kt

@@ -0,0 +1,25 @@
+package ac.uk.ebi.biostd.handlers.exception
+
+import ac.uk.ebi.biostd.handlers.common.HANDLERS_SUBSYSTEM
+import ac.uk.ebi.biostd.handlers.common.SYSTEM_NAME
+import ebi.ac.uk.commons.http.slack.Alert
+import ebi.ac.uk.commons.http.slack.NotificationsSender
+import kotlinx.coroutines.runBlocking
+import org.springframework.amqp.AmqpRejectAndDontRequeueException
+import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler
+
+class CustomErrorHandler(
+    private val notificationsSender: NotificationsSender,
+) : ConditionalRejectingErrorHandler() {
+    override fun handleError(t: Throwable) {
+        runBlocking {
+            notificationsSender.send(Alert(SYSTEM_NAME, HANDLERS_SUBSYSTEM, "$ERROR_MESSAGE: ${t.localizedMessage}"))
+        }
+
+        throw AmqpRejectAndDontRequeueException(t)
+    }
+
+    companion object {
+        internal const val ERROR_MESSAGE = "Problem processing rabbit message for notification"
+    }
+}

submission/submission-handlers/src/test/kotlin/ac/uk/ebi/biostd/handlers/exception/CustomErrorHandlerTest.kt

@@ -0,0 +1,41 @@
+package ac.uk.ebi.biostd.handlers.exception
+
+import ac.uk.ebi.biostd.handlers.common.HANDLERS_SUBSYSTEM
+import ac.uk.ebi.biostd.handlers.common.SYSTEM_NAME
+import ac.uk.ebi.biostd.handlers.exception.CustomErrorHandler.Companion.ERROR_MESSAGE
+import ebi.ac.uk.commons.http.slack.Alert
+import ebi.ac.uk.commons.http.slack.NotificationsSender
+import io.mockk.coEvery
+import io.mockk.coVerify
+import io.mockk.every
+import io.mockk.impl.annotations.MockK
+import io.mockk.junit5.MockKExtension
+import io.mockk.slot
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.extension.ExtendWith
+import org.springframework.amqp.AmqpRejectAndDontRequeueException
+import kotlin.test.assertFailsWith
+
+@ExtendWith(MockKExtension::class)
+class CustomErrorHandlerTest(
+    @MockK private val throwable: Throwable,
+    @MockK private val notificationsSender: NotificationsSender,
+) {
+    private val testInstance = CustomErrorHandler(notificationsSender)
+
+    @Test
+    fun `handle error`() {
+        val alertSlot = slot<Alert>()
+        val expectedAlert = Alert(SYSTEM_NAME, HANDLERS_SUBSYSTEM, "$ERROR_MESSAGE: the error message")
+
+        every { throwable.localizedMessage } returns "the error message"
+        coEvery { notificationsSender.send(capture(alertSlot)) } answers { nothing }
+
+        assertFailsWith<AmqpRejectAndDontRequeueException> { testInstance.handleError(throwable) }
+
+        val alert = alertSlot.captured
+        assertThat(alert).isEqualToComparingFieldByField(expectedAlert)
+        coVerify(exactly = 1) { notificationsSender.send(alert) }
+    }
+}