Merge pull request #45 from CCC-MF/issue_34

Verwendung einer applikationsweiten Retry-Konfiguration
2025-04-19 17:26:51 +00:00 · 2024-02-29 08:51:28 +01:00 · 2024-02-29 08:51:28 +01:00 · 46ddaf10f7
commit 46ddaf10f7
parent 25f286f73b 408b121f26
8 changed files with 196 additions and 90 deletions
--- a/src/integrationTest/kotlin/dev/dnpm/etl/processor/config/AppConfigurationTest.kt
+++ b/src/integrationTest/kotlin/dev/dnpm/etl/processor/config/AppConfigurationTest.kt
@ -20,6 +20,7 @@
 package dev.dnpm.etl.processor.config

 import com.fasterxml.jackson.databind.ObjectMapper
+import dev.dnpm.etl.processor.input.KafkaInputListener
 import dev.dnpm.etl.processor.monitoring.RequestRepository
 import dev.dnpm.etl.processor.output.KafkaMtbFileSender
 import dev.dnpm.etl.processor.output.RestMtbFileSender
@ -37,6 +38,7 @@ import org.springframework.boot.test.context.SpringBootTest
 import org.springframework.boot.test.mock.mockito.MockBean
 import org.springframework.boot.test.mock.mockito.MockBeans
 import org.springframework.context.ApplicationContext
+import org.springframework.retry.support.RetryTemplate
 import org.springframework.security.crypto.password.PasswordEncoder
 import org.springframework.security.provisioning.InMemoryUserDetailsManager
 import org.springframework.test.context.ContextConfiguration
@ -78,8 +80,8 @@ class AppConfigurationTest {
    @TestPropertySource(
        properties = [
            "app.kafka.servers=localhost:9092",
-            "app.kafka.topic=test",
-            "app.kafka.response-topic=test-response",
+            "app.kafka.output-topic=test",
+            "app.kafka.output-response-topic=test-response",
            "app.kafka.group-id=test"
        ]
    )
@ -99,8 +101,8 @@ class AppConfigurationTest {
        properties = [
            "app.rest.uri=http://localhost:9000",
            "app.kafka.servers=localhost:9092",
-            "app.kafka.topic=test",
-            "app.kafka.response-topic=test-response",
+            "app.kafka.output-topic=test",
+            "app.kafka.output-response-topic=test-response",
            "app.kafka.group-id=test"
        ]
    )
@ -114,6 +116,43 @@ class AppConfigurationTest {

    }

+    @Nested
+    @TestPropertySource(
+        properties = [
+            "app.kafka.servers=localhost:9092",
+            "app.kafka.output-topic=test",
+            "app.kafka.output-response-topic=test-response",
+            "app.kafka.group-id=test"
+        ]
+    )
+    inner class AppConfigurationWithoutKafkaInputTest(private val context: ApplicationContext) {
+
+        @Test
+        fun shouldNotUseKafkaInputListener() {
+            assertThrows<NoSuchBeanDefinitionException> { context.getBean(KafkaInputListener::class.java) }
+        }
+
+    }
+
+    @Nested
+    @TestPropertySource(
+        properties = [
+            "app.kafka.servers=localhost:9092",
+            "app.kafka.input-topic=test_input",
+            "app.kafka.output-topic=test",
+            "app.kafka.output-response-topic=test-response",
+            "app.kafka.group-id=test"
+        ]
+    )
+    inner class AppConfigurationUsingKafkaInputTest(private val context: ApplicationContext) {
+
+        @Test
+        fun shouldUseKafkaInputListener() {
+            assertThat(context.getBean(KafkaInputListener::class.java)).isNotNull
+        }
+
+    }
+
    @Nested
    @TestPropertySource(
        properties = [
@ -238,4 +277,30 @@ class AppConfigurationTest {

    }

+    @Nested
+    @TestPropertySource(
+        properties = [
+            "app.rest.uri=http://localhost:9000",
+            "app.max-retry-attempts=5"
+        ]
+    )
+    inner class AppConfigurationRetryTest(private val context: ApplicationContext) {
+
+        private val maxRetryAttempts = 5
+
+        @Test
+        fun shouldUseRetryTemplateWithConfiguredMaxAttempts() {
+            val retryTemplate = context.getBean(RetryTemplate::class.java)
+            assertThat(retryTemplate).isNotNull
+
+            assertThrows<RuntimeException> {
+                retryTemplate.execute<Void, RuntimeException> {
+                    assertThat(it.retryCount).isLessThan(maxRetryAttempts)
+                    throw RuntimeException()
+                }
+            }
+        }
+
+    }
+
 }
--- a/src/main/java/dev/dnpm/etl/processor/pseudonym/GpasPseudonymGenerator.java
+++ b/src/main/java/dev/dnpm/etl/processor/pseudonym/GpasPseudonymGenerator.java
@ -22,21 +22,6 @@ package dev.dnpm.etl.processor.pseudonym;
 import ca.uhn.fhir.context.FhirContext;
 import ca.uhn.fhir.parser.IParser;
 import dev.dnpm.etl.processor.config.GPasConfigProperties;
-import java.io.BufferedInputStream;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.net.ConnectException;
-import java.security.KeyManagementException;
-import java.security.KeyStore;
-import java.security.KeyStoreException;
-import java.security.NoSuchAlgorithmException;
-import java.security.cert.CertificateException;
-import java.security.cert.CertificateFactory;
-import java.security.cert.X509Certificate;
-import java.util.Base64;
-import java.util.HashMap;
-import javax.net.ssl.SSLContext;
-import javax.net.ssl.TrustManagerFactory;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.apache.hc.client5.http.impl.classic.HttpClients;
@ -54,35 +39,39 @@ import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.http.HttpEntity;
-import org.springframework.http.HttpHeaders;
-import org.springframework.http.HttpMethod;
-import org.springframework.http.MediaType;
-import org.springframework.http.ResponseEntity;
+import org.springframework.http.*;
 import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
-import org.springframework.retry.RetryCallback;
-import org.springframework.retry.RetryContext;
-import org.springframework.retry.RetryListener;
-import org.springframework.retry.RetryPolicy;
-import org.springframework.retry.backoff.ExponentialBackOffPolicy;
-import org.springframework.retry.policy.SimpleRetryPolicy;
 import org.springframework.retry.support.RetryTemplate;
-import org.springframework.web.client.RestClientException;
 import org.springframework.web.client.RestTemplate;

+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManagerFactory;
+import java.io.BufferedInputStream;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.security.KeyManagementException;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.cert.CertificateException;
+import java.security.cert.CertificateFactory;
+import java.security.cert.X509Certificate;
+import java.util.Base64;
+
 public class GpasPseudonymGenerator implements Generator {

    private final static FhirContext r4Context = FhirContext.forR4();
    private final String gPasUrl;
    private final String psnTargetDomain;
    private final HttpHeaders httpHeader;
-    private final RetryTemplate retryTemplate = defaultTemplate();
+    private final RetryTemplate retryTemplate;
    private final Logger log = LoggerFactory.getLogger(GpasPseudonymGenerator.class);

    private SSLContext customSslContext;
    private RestTemplate restTemplate;

-    public GpasPseudonymGenerator(GPasConfigProperties gpasCfg) {
+    public GpasPseudonymGenerator(GPasConfigProperties gpasCfg, RetryTemplate retryTemplate) {
+        this.retryTemplate = retryTemplate;

        this.gPasUrl = gpasCfg.getUri();
        this.psnTargetDomain = gpasCfg.getTarget();
@ -202,31 +191,6 @@ public class GpasPseudonymGenerator implements Generator {
        return headers;
    }

-    protected RetryTemplate defaultTemplate() {
-        RetryTemplate retryTemplate = new RetryTemplate();
-        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
-        backOffPolicy.setInitialInterval(1000);
-        backOffPolicy.setMultiplier(1.25);
-        retryTemplate.setBackOffPolicy(backOffPolicy);
-        HashMap<Class<? extends Throwable>, Boolean> retryableExceptions = new HashMap<>();
-        retryableExceptions.put(RestClientException.class, true);
-        retryableExceptions.put(ConnectException.class, true);
-        RetryPolicy retryPolicy = new SimpleRetryPolicy(3, retryableExceptions);
-        retryTemplate.setRetryPolicy(retryPolicy);
-
-        retryTemplate.registerListener(new RetryListener() {
-            @Override
-            public <T, E extends Throwable> void onError(RetryContext context,
-                RetryCallback<T, E> callback, Throwable throwable) {
-                log.warn("HTTP Error occurred: {}. Retrying {}", throwable.getMessage(),
-                    context.getRetryCount());
-                RetryListener.super.onError(context, callback, throwable);
-            }
-        });
-
-        return retryTemplate;
-    }
-
    /**
     * Read SSL root certificate and return SSLContext
     *
--- a/src/main/kotlin/dev/dnpm/etl/processor/config/AppConfigProperties.kt
+++ b/src/main/kotlin/dev/dnpm/etl/processor/config/AppConfigProperties.kt
@ -30,7 +30,8 @@ data class AppConfigProperties(
        replacement = "app.pseudonymize.generator"
    )
    var pseudonymizer: PseudonymGenerator = PseudonymGenerator.BUILDIN,
-    var transformations: List<TransformationProperties> = listOf()
+    var transformations: List<TransformationProperties> = listOf(),
+    var maxRetryAttempts: Int = 3
 ) {
    companion object {
        const val NAME = "app"
@ -72,10 +73,21 @@ data class RestTargetProperties(
    }
 }

-@ConfigurationProperties(KafkaTargetProperties.NAME)
-data class KafkaTargetProperties(
-    val topic: String = "etl-processor",
-    val responseTopic: String = "${topic}_response",
+@ConfigurationProperties(KafkaProperties.NAME)
+data class KafkaProperties(
+    val inputTopic: String?,
+    val outputTopic: String = "etl-processor",
+    @get:DeprecatedConfigurationProperty(
+        reason = "Deprecated",
+        replacement = "outputTopic"
+    )
+    val topic: String = outputTopic,
+    val outputResponseTopic: String = "${outputTopic}_response",
+    @get:DeprecatedConfigurationProperty(
+        reason = "Deprecated",
+        replacement = "outputResponseTopic"
+    )
+    val responseTopic: String = outputResponseTopic,
    val groupId: String = "${topic}_group",
    val servers: String = ""
 ) {
--- a/src/main/kotlin/dev/dnpm/etl/processor/config/AppConfiguration.kt
+++ b/src/main/kotlin/dev/dnpm/etl/processor/config/AppConfiguration.kt
@ -35,13 +35,15 @@ 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.RetryCallback
+import org.springframework.retry.RetryContext
+import org.springframework.retry.RetryListener
 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 org.springframework.security.crypto.password.PasswordEncoder
 import org.springframework.security.provisioning.InMemoryUserDetailsManager
-import org.springframework.security.provisioning.UserDetailsManager
 import reactor.core.publisher.Sinks
 import kotlin.time.Duration.Companion.seconds
 import kotlin.time.toJavaDuration
@ -62,8 +64,8 @@ class AppConfiguration {

    @ConditionalOnProperty(value = ["app.pseudonymize.generator"], havingValue = "GPAS")
    @Bean
-    fun gpasPseudonymGenerator(configProperties: GPasConfigProperties): Generator {
-        return GpasPseudonymGenerator(configProperties)
+    fun gpasPseudonymGenerator(configProperties: GPasConfigProperties, retryTemplate: RetryTemplate): Generator {
+        return GpasPseudonymGenerator(configProperties, retryTemplate)
    }

    @ConditionalOnProperty(value = ["app.pseudonymize.generator"], havingValue = "BUILDIN", matchIfMissing = true)
@ -75,8 +77,8 @@ class AppConfiguration {
    @ConditionalOnProperty(value = ["app.pseudonymizer"], havingValue = "GPAS")
    @ConditionalOnMissingBean
    @Bean
-    fun gpasPseudonymGeneratorOnDeprecatedProperty(configProperties: GPasConfigProperties): Generator {
-        return GpasPseudonymGenerator(configProperties)
+    fun gpasPseudonymGeneratorOnDeprecatedProperty(configProperties: GPasConfigProperties, retryTemplate: RetryTemplate): Generator {
+        return GpasPseudonymGenerator(configProperties, retryTemplate)
    }

    @ConditionalOnProperty(value = ["app.pseudonymizer"], havingValue = "BUILDIN")
@ -111,11 +113,20 @@ class AppConfiguration {
    }

    @Bean
-    fun retryTemplate(): RetryTemplate {
+    fun retryTemplate(configProperties: AppConfigProperties): RetryTemplate {
        return RetryTemplateBuilder()
            .notRetryOn(IllegalArgumentException::class.java)
-            .fixedBackoff(5.seconds.toJavaDuration())
-            .customPolicy(SimpleRetryPolicy(3))
+            .exponentialBackoff(2.seconds.toJavaDuration(), 1.25, 5.seconds.toJavaDuration())
+            .customPolicy(SimpleRetryPolicy(configProperties.maxRetryAttempts))
+            .withListener(object : RetryListener {
+                override fun <T : Any, E : Throwable> onError(
+                    context: RetryContext,
+                    callback: RetryCallback<T, E>,
+                    throwable: Throwable
+                ) {
+                    logger.warn("Error occured: {}. Retrying {}", throwable.message, context.retryCount)
+                }
+            })
            .build()
    }

--- a/src/main/kotlin/dev/dnpm/etl/processor/config/AppKafkaConfiguration.kt
+++ b/src/main/kotlin/dev/dnpm/etl/processor/config/AppKafkaConfiguration.kt
@ -20,6 +20,7 @@
 package dev.dnpm.etl.processor.config

 import com.fasterxml.jackson.databind.ObjectMapper
+import dev.dnpm.etl.processor.input.KafkaInputListener
 import dev.dnpm.etl.processor.monitoring.ConnectionCheckService
 import dev.dnpm.etl.processor.monitoring.KafkaConnectionCheckService
 import dev.dnpm.etl.processor.output.KafkaMtbFileSender
@ -42,9 +43,9 @@ import reactor.core.publisher.Sinks

@Configuration
@EnableConfigurationProperties(
-    value = [KafkaTargetProperties::class]
+    value = [KafkaProperties::class]
 )
-@ConditionalOnProperty(value = ["app.kafka.topic", "app.kafka.servers"])
+@ConditionalOnProperty(value = ["app.kafka.servers"])
@ConditionalOnMissingBean(MtbFileSender::class)
@Order(-5)
 class AppKafkaConfiguration {
@ -54,21 +55,21 @@ class AppKafkaConfiguration {
    @Bean
    fun kafkaMtbFileSender(
        kafkaTemplate: KafkaTemplate<String, String>,
-        kafkaTargetProperties: KafkaTargetProperties,
+        kafkaProperties: KafkaProperties,
        retryTemplate: RetryTemplate,
        objectMapper: ObjectMapper
    ): MtbFileSender {
        logger.info("Selected 'KafkaMtbFileSender'")
-        return KafkaMtbFileSender(kafkaTemplate, kafkaTargetProperties, retryTemplate, objectMapper)
+        return KafkaMtbFileSender(kafkaTemplate, kafkaProperties, retryTemplate, objectMapper)
    }

    @Bean
-    fun kafkaListenerContainer(
+    fun kafkaResponseListenerContainer(
        consumerFactory: ConsumerFactory<String, String>,
-        kafkaTargetProperties: KafkaTargetProperties,
+        kafkaProperties: KafkaProperties,
        kafkaResponseProcessor: KafkaResponseProcessor
    ): KafkaMessageListenerContainer<String, String> {
-        val containerProperties = ContainerProperties(kafkaTargetProperties.responseTopic)
+        val containerProperties = ContainerProperties(kafkaProperties.responseTopic)
        containerProperties.messageListener = kafkaResponseProcessor
        return KafkaMessageListenerContainer(consumerFactory, containerProperties)
    }
@ -81,6 +82,26 @@ class AppKafkaConfiguration {
        return KafkaResponseProcessor(applicationEventPublisher, objectMapper)
    }

+    @Bean
+    @ConditionalOnProperty(value = ["app.kafka.input-topic"])
+    fun kafkaInputListenerContainer(
+        consumerFactory: ConsumerFactory<String, String>,
+        kafkaProperties: KafkaProperties,
+        kafkaInputListener: KafkaInputListener
+    ): KafkaMessageListenerContainer<String, String> {
+        val containerProperties = ContainerProperties(kafkaProperties.inputTopic)
+        containerProperties.messageListener = kafkaInputListener
+        return KafkaMessageListenerContainer(consumerFactory, containerProperties)
+    }
+
+    @Bean
+    @ConditionalOnProperty(value = ["app.kafka.input-topic"])
+    fun kafkaInputListener(
+        applicationEventPublisher: ApplicationEventPublisher,
+    ): KafkaInputListener {
+        return KafkaInputListener(applicationEventPublisher)
+    }
+
    @Bean
    fun connectionCheckService(consumerFactory: ConsumerFactory<String, String>, configsUpdateProducer: Sinks.Many<Boolean>): ConnectionCheckService {
        return KafkaConnectionCheckService(consumerFactory.createConsumer(), configsUpdateProducer)
--- a/src/main/kotlin/dev/dnpm/etl/processor/input/KafkaInputListener.kt
+++ b/src/main/kotlin/dev/dnpm/etl/processor/input/KafkaInputListener.kt
@ -0,0 +1,33 @@
+/*
+ * This file is part of ETL-Processor
+ *
+ * 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
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package dev.dnpm.etl.processor.input
+
+import de.ukw.ccc.bwhc.dto.MtbFile
+import org.apache.kafka.clients.consumer.ConsumerRecord
+import org.springframework.context.ApplicationEventPublisher
+import org.springframework.kafka.listener.MessageListener
+
+class KafkaInputListener(
+    private val applicationEventPublisher: ApplicationEventPublisher
+) : MessageListener<String, MtbFile> {
+    override fun onMessage(data: ConsumerRecord<String, MtbFile>) {
+        TODO("Not yet implemented")
+    }
+}
--- a/src/main/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSender.kt
+++ b/src/main/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSender.kt
@ -22,7 +22,7 @@ package dev.dnpm.etl.processor.output
 import com.fasterxml.jackson.databind.ObjectMapper
 import de.ukw.ccc.bwhc.dto.Consent
 import de.ukw.ccc.bwhc.dto.MtbFile
-import dev.dnpm.etl.processor.config.KafkaTargetProperties
+import dev.dnpm.etl.processor.config.KafkaProperties
 import dev.dnpm.etl.processor.monitoring.RequestStatus
 import org.slf4j.LoggerFactory
 import org.springframework.kafka.core.KafkaTemplate
@ -30,7 +30,7 @@ import org.springframework.retry.support.RetryTemplate

 class KafkaMtbFileSender(
    private val kafkaTemplate: KafkaTemplate<String, String>,
-    private val kafkaTargetProperties: KafkaTargetProperties,
+    private val kafkaProperties: KafkaProperties,
    private val retryTemplate: RetryTemplate,
    private val objectMapper: ObjectMapper
 ) : MtbFileSender {
@ -41,7 +41,7 @@ class KafkaMtbFileSender(
        return try {
            return retryTemplate.execute<MtbFileSender.Response, Exception> {
                val result = kafkaTemplate.send(
-                    kafkaTargetProperties.topic,
+                    kafkaProperties.topic,
                    key(request),
                    objectMapper.writeValueAsString(Data(request.requestId, request.mtbFile))
                )
@ -71,7 +71,7 @@ class KafkaMtbFileSender(
        return try {
            return retryTemplate.execute<MtbFileSender.Response, Exception> {
                val result = kafkaTemplate.send(
-                    kafkaTargetProperties.topic,
+                    kafkaProperties.topic,
                    key(request),
                    objectMapper.writeValueAsString(Data(request.requestId, dummyMtbFile))
                )
@ -90,7 +90,7 @@ class KafkaMtbFileSender(
    }

    override fun endpoint(): String {
-        return "${this.kafkaTargetProperties.servers} (${this.kafkaTargetProperties.topic}/${this.kafkaTargetProperties.responseTopic})"
+        return "${this.kafkaProperties.servers} (${this.kafkaProperties.topic}/${this.kafkaProperties.responseTopic})"
    }

    private fun key(request: MtbFileSender.MtbFileRequest): String {
--- a/src/test/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSenderTest.kt
+++ b/src/test/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSenderTest.kt
@ -21,7 +21,7 @@ package dev.dnpm.etl.processor.output

 import com.fasterxml.jackson.databind.ObjectMapper
 import de.ukw.ccc.bwhc.dto.*
-import dev.dnpm.etl.processor.config.KafkaTargetProperties
+import dev.dnpm.etl.processor.config.KafkaProperties
 import dev.dnpm.etl.processor.monitoring.RequestStatus
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.BeforeEach
@ -53,13 +53,13 @@ class KafkaMtbFileSenderTest {
    fun setup(
        @Mock kafkaTemplate: KafkaTemplate<String, String>
    ) {
-        val kafkaTargetProperties = KafkaTargetProperties("testtopic")
+        val kafkaProperties = KafkaProperties("testtopic")
        val retryTemplate = RetryTemplateBuilder().customPolicy(SimpleRetryPolicy(1)).build()

        this.objectMapper = ObjectMapper()
        this.kafkaTemplate = kafkaTemplate

-        this.kafkaMtbFileSender = KafkaMtbFileSender(kafkaTemplate, kafkaTargetProperties, retryTemplate, objectMapper)
+        this.kafkaMtbFileSender = KafkaMtbFileSender(kafkaTemplate, kafkaProperties, retryTemplate, objectMapper)
    }

    @ParameterizedTest
@ -125,9 +125,9 @@ class KafkaMtbFileSenderTest {
    @ParameterizedTest
    @MethodSource("requestWithResponseSource")
    fun shouldRetryOnMtbFileKafkaSendError(testData: TestData) {
-        val kafkaTargetProperties = KafkaTargetProperties("testtopic")
+        val kafkaProperties = KafkaProperties("testtopic")
        val retryTemplate = RetryTemplateBuilder().customPolicy(SimpleRetryPolicy(3)).build()
-        this.kafkaMtbFileSender = KafkaMtbFileSender(this.kafkaTemplate, kafkaTargetProperties, retryTemplate, this.objectMapper)
+        this.kafkaMtbFileSender = KafkaMtbFileSender(this.kafkaTemplate, kafkaProperties, retryTemplate, this.objectMapper)

        doAnswer {
            if (null != testData.exception) {
@ -151,9 +151,9 @@ class KafkaMtbFileSenderTest {
    @ParameterizedTest
    @MethodSource("requestWithResponseSource")
    fun shouldRetryOnDeleteKafkaSendError(testData: TestData) {
-        val kafkaTargetProperties = KafkaTargetProperties("testtopic")
+        val kafkaProperties = KafkaProperties("testtopic")
        val retryTemplate = RetryTemplateBuilder().customPolicy(SimpleRetryPolicy(3)).build()
-        this.kafkaMtbFileSender = KafkaMtbFileSender(this.kafkaTemplate, kafkaTargetProperties, retryTemplate, this.objectMapper)
+        this.kafkaMtbFileSender = KafkaMtbFileSender(this.kafkaTemplate, kafkaProperties, retryTemplate, this.objectMapper)

        doAnswer {
            if (null != testData.exception) {