feat #17: initial support for request retry

src/main/kotlin/dev/dnpm/etl/processor/config/AppConfiguration.kt

@@ -32,8 +32,14 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
+import org.springframework.retry.policy.SimpleRetryPolicy
+import org.springframework.retry.support.RetryTemplate
+import org.springframework.retry.support.RetryTemplateBuilder
 import org.springframework.scheduling.annotation.EnableScheduling
 import reactor.core.publisher.Sinks
+import kotlin.time.Duration.Companion.seconds
+import kotlin.time.toJavaDuration
+
 
 @Configuration
 @EnableConfigurationProperties(
@@ -89,5 +95,14 @@ class AppConfiguration {
         })
     }
 
+    @Bean
+    fun retryTemplate(): RetryTemplate {
+        return RetryTemplateBuilder()
+            .notRetryOn(IllegalArgumentException::class.java)
+            .fixedBackoff(5.seconds.toJavaDuration())
+            .customPolicy(SimpleRetryPolicy(3))
+            .build()
+    }
+
 }
 

src/main/kotlin/dev/dnpm/etl/processor/config/AppKafkaConfiguration.kt

@@ -37,6 +37,7 @@ import org.springframework.kafka.core.ConsumerFactory
 import org.springframework.kafka.core.KafkaTemplate
 import org.springframework.kafka.listener.ContainerProperties
 import org.springframework.kafka.listener.KafkaMessageListenerContainer
+import org.springframework.retry.support.RetryTemplate
 
 @Configuration
 @EnableConfigurationProperties(
@@ -53,10 +54,11 @@ class AppKafkaConfiguration {
     fun kafkaMtbFileSender(
         kafkaTemplate: KafkaTemplate<String, String>,
         kafkaTargetProperties: KafkaTargetProperties,
+        retryTemplate: RetryTemplate,
         objectMapper: ObjectMapper
     ): MtbFileSender {
         logger.info("Selected 'KafkaMtbFileSender'")
-        return KafkaMtbFileSender(kafkaTemplate, kafkaTargetProperties, objectMapper)
+        return KafkaMtbFileSender(kafkaTemplate, kafkaTargetProperties, retryTemplate, objectMapper)
     }
 
     @Bean

src/main/kotlin/dev/dnpm/etl/processor/config/AppRestConfiguration.kt

@@ -30,6 +30,7 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
 import org.springframework.core.annotation.Order
+import org.springframework.retry.support.RetryTemplate
 import org.springframework.web.client.RestTemplate
 
 @Configuration
@@ -51,9 +52,13 @@ class AppRestConfiguration {
     }
 
     @Bean
-    fun restMtbFileSender(restTemplate: RestTemplate, restTargetProperties: RestTargetProperties): MtbFileSender {
+    fun restMtbFileSender(
+        restTemplate: RestTemplate,
+        restTargetProperties: RestTargetProperties,
+        retryTemplate: RetryTemplate
+    ): MtbFileSender {
         logger.info("Selected 'RestMtbFileSender'")
-        return RestMtbFileSender(restTemplate, restTargetProperties)
+        return RestMtbFileSender(restTemplate, restTargetProperties, retryTemplate)
     }
 
     @Bean

src/main/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSender.kt

@@ -1,7 +1,7 @@
 /*
  * This file is part of ETL-Processor
  *
- * Copyright (c) 2023  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ * Copyright (c) 2024  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Affero General Public License as published
@@ -26,10 +26,12 @@ import dev.dnpm.etl.processor.config.KafkaTargetProperties
 import dev.dnpm.etl.processor.monitoring.RequestStatus
 import org.slf4j.LoggerFactory
 import org.springframework.kafka.core.KafkaTemplate
+import org.springframework.retry.support.RetryTemplate
 
 class KafkaMtbFileSender(
     private val kafkaTemplate: KafkaTemplate<String, String>,
     private val kafkaTargetProperties: KafkaTargetProperties,
+    private val retryTemplate: RetryTemplate,
     private val objectMapper: ObjectMapper
 ) : MtbFileSender {
 
@@ -37,16 +39,18 @@ class KafkaMtbFileSender(
 
     override fun send(request: MtbFileSender.MtbFileRequest): MtbFileSender.Response {
         return try {
-            val result = kafkaTemplate.send(
+            return retryTemplate.execute<MtbFileSender.Response, Exception> {
-                kafkaTargetProperties.topic,
+                val result = kafkaTemplate.send(
-                key(request),
+                    kafkaTargetProperties.topic,
-                objectMapper.writeValueAsString(Data(request.requestId, request.mtbFile))
+                    key(request),
-            )
+                    objectMapper.writeValueAsString(Data(request.requestId, request.mtbFile))
-            if (result.get() != null) {
+                )
-                logger.debug("Sent file via KafkaMtbFileSender")
+                if (result.get() != null) {
-                MtbFileSender.Response(RequestStatus.UNKNOWN)
+                    logger.debug("Sent file via KafkaMtbFileSender")
-            } else {
+                    MtbFileSender.Response(RequestStatus.UNKNOWN)
-                MtbFileSender.Response(RequestStatus.ERROR)
+                } else {
+                    MtbFileSender.Response(RequestStatus.ERROR)
+                }
             }
         } catch (e: Exception) {
             logger.error("An error occurred sending to kafka", e)
@@ -65,17 +69,19 @@ class KafkaMtbFileSender(
             .build()
 
         return try {
-            val result = kafkaTemplate.send(
+            return retryTemplate.execute<MtbFileSender.Response, Exception> {
-                kafkaTargetProperties.topic,
+                val result = kafkaTemplate.send(
-                key(request),
+                    kafkaTargetProperties.topic,
-                objectMapper.writeValueAsString(Data(request.requestId, dummyMtbFile))
+                    key(request),
-            )
+                    objectMapper.writeValueAsString(Data(request.requestId, dummyMtbFile))
+                )
 
-            if (result.get() != null) {
+                if (result.get() != null) {
-                logger.debug("Sent deletion request via KafkaMtbFileSender")
+                    logger.debug("Sent deletion request via KafkaMtbFileSender")
-                MtbFileSender.Response(RequestStatus.UNKNOWN)
+                    MtbFileSender.Response(RequestStatus.UNKNOWN)
-            } else {
+                } else {
-                MtbFileSender.Response(RequestStatus.ERROR)
+                    MtbFileSender.Response(RequestStatus.ERROR)
+                }
             }
         } catch (e: Exception) {
             logger.error("An error occurred sending to kafka", e)

src/main/kotlin/dev/dnpm/etl/processor/output/RestMtbFileSender.kt

@@ -1,7 +1,7 @@
 /*
  * This file is part of ETL-Processor
  *
- * Copyright (c) 2023  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ * Copyright (c) 2024  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Affero General Public License as published
@@ -25,32 +25,39 @@ import org.slf4j.LoggerFactory
 import org.springframework.http.HttpEntity
 import org.springframework.http.HttpHeaders
 import org.springframework.http.MediaType
+import org.springframework.retry.support.RetryTemplate
 import org.springframework.web.client.RestClientException
 import org.springframework.web.client.RestTemplate
 
 class RestMtbFileSender(
     private val restTemplate: RestTemplate,
-    private val restTargetProperties: RestTargetProperties
+    private val restTargetProperties: RestTargetProperties,
+    private val retryTemplate: RetryTemplate
 ) : MtbFileSender {
 
     private val logger = LoggerFactory.getLogger(RestMtbFileSender::class.java)
 
     override fun send(request: MtbFileSender.MtbFileRequest): MtbFileSender.Response {
         try {
-            val headers = HttpHeaders()
+            return retryTemplate.execute<MtbFileSender.Response, Exception> {
-            headers.contentType = MediaType.APPLICATION_JSON
+                val headers = HttpHeaders()
-            val entityReq = HttpEntity(request.mtbFile, headers)
+                headers.contentType = MediaType.APPLICATION_JSON
-            val response = restTemplate.postForEntity(
+                val entityReq = HttpEntity(request.mtbFile, headers)
-                "${restTargetProperties.uri}/MTBFile",
+                val response = restTemplate.postForEntity(
-                entityReq,
+                    "${restTargetProperties.uri}/MTBFile",
-                String::class.java
+                    entityReq,
-            )
+                    String::class.java
-            if (!response.statusCode.is2xxSuccessful) {
+                )
-                logger.warn("Error sending to remote system: {}", response.body)
+                if (!response.statusCode.is2xxSuccessful) {
-                return MtbFileSender.Response(response.statusCode.asRequestStatus(), "Status-Code: ${response.statusCode.value()}")
+                    logger.warn("Error sending to remote system: {}", response.body)
+                    return@execute MtbFileSender.Response(
+                        response.statusCode.asRequestStatus(),
+                        "Status-Code: ${response.statusCode.value()}"
+                    )
+                }
+                logger.debug("Sent file via RestMtbFileSender")
+                return@execute MtbFileSender.Response(response.statusCode.asRequestStatus(), response.body.orEmpty())
             }
-            logger.debug("Sent file via RestMtbFileSender")
-            return MtbFileSender.Response(response.statusCode.asRequestStatus(), response.body.orEmpty())
         } catch (e: IllegalArgumentException) {
             logger.error("Not a valid URI to export to: '{}'", restTargetProperties.uri!!)
         } catch (e: RestClientException) {
@@ -62,16 +69,18 @@ class RestMtbFileSender(
 
     override fun send(request: MtbFileSender.DeleteRequest): MtbFileSender.Response {
         try {
-            val headers = HttpHeaders()
+            return retryTemplate.execute<MtbFileSender.Response, Exception> {
-            headers.contentType = MediaType.APPLICATION_JSON
+                val headers = HttpHeaders()
-            val entityReq = HttpEntity(null, headers)
+                headers.contentType = MediaType.APPLICATION_JSON
-            restTemplate.delete(
+                val entityReq = HttpEntity(null, headers)
-                "${restTargetProperties.uri}/Patient/${request.patientId}",
+                restTemplate.delete(
-                entityReq,
+                    "${restTargetProperties.uri}/Patient/${request.patientId}",
-                String::class.java
+                    entityReq,
-            )
+                    String::class.java
-            logger.debug("Sent file via RestMtbFileSender")
+                )
-            return MtbFileSender.Response(RequestStatus.SUCCESS)
+                logger.debug("Sent file via RestMtbFileSender")
+                return@execute MtbFileSender.Response(RequestStatus.SUCCESS)
+            }
         } catch (e: IllegalArgumentException) {
             logger.error("Not a valid URI to export to: '{}'", restTargetProperties.uri!!)
         } catch (e: RestClientException) {

src/test/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSenderTest.kt

@@ -1,7 +1,7 @@
 /*
  * This file is part of ETL-Processor
  *
- * Copyright (c) 2023  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ * Copyright (c) 2024  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Affero General Public License as published
@@ -35,6 +35,8 @@ import org.mockito.junit.jupiter.MockitoExtension
 import org.mockito.kotlin.*
 import org.springframework.kafka.core.KafkaTemplate
 import org.springframework.kafka.support.SendResult
+import org.springframework.retry.policy.SimpleRetryPolicy
+import org.springframework.retry.support.RetryTemplateBuilder
 import java.util.concurrent.CompletableFuture.completedFuture
 import java.util.concurrent.ExecutionException
 
@@ -52,10 +54,12 @@ class KafkaMtbFileSenderTest {
         @Mock kafkaTemplate: KafkaTemplate<String, String>
     ) {
         val kafkaTargetProperties = KafkaTargetProperties("testtopic")
+        val retryTemplate = RetryTemplateBuilder().customPolicy(SimpleRetryPolicy(1)).build()
+
         this.objectMapper = ObjectMapper()
         this.kafkaTemplate = kafkaTemplate
 
-        this.kafkaMtbFileSender = KafkaMtbFileSender(kafkaTemplate, kafkaTargetProperties, objectMapper)
+        this.kafkaMtbFileSender = KafkaMtbFileSender(kafkaTemplate, kafkaTargetProperties, retryTemplate, objectMapper)
     }
 
     @ParameterizedTest

src/test/kotlin/dev/dnpm/etl/processor/output/RestMtbFileSenderTest.kt

@@ -1,7 +1,7 @@
 /*
  * This file is part of ETL-Processor
  *
- * Copyright (c) 2023  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ * Copyright (c) 2024  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Affero General Public License as published
@@ -28,6 +28,8 @@ import org.junit.jupiter.params.ParameterizedTest
 import org.junit.jupiter.params.provider.MethodSource
 import org.springframework.http.HttpMethod
 import org.springframework.http.HttpStatus
+import org.springframework.retry.policy.SimpleRetryPolicy
+import org.springframework.retry.support.RetryTemplateBuilder
 import org.springframework.test.web.client.MockRestServiceServer
 import org.springframework.test.web.client.match.MockRestRequestMatchers.method
 import org.springframework.test.web.client.match.MockRestRequestMatchers.requestTo
@@ -44,10 +46,11 @@ class RestMtbFileSenderTest {
     fun setup() {
         val restTemplate = RestTemplate()
         val restTargetProperties = RestTargetProperties("http://localhost:9000/mtbfile")
+        val retryTemplate = RetryTemplateBuilder().customPolicy(SimpleRetryPolicy(1)).build()
 
         this.mockRestServiceServer = MockRestServiceServer.createServer(restTemplate)
 
-        this.restMtbFileSender = RestMtbFileSender(restTemplate, restTargetProperties)
+        this.restMtbFileSender = RestMtbFileSender(restTemplate, restTargetProperties, retryTemplate)
     }
 
     @ParameterizedTest