Fix error 500 while pulling test container cp-kafka

Issue-ID: CPS-1492
Change-Id: I76be5264b111be36f9d2d1f0af13606244189e7c
Signed-off-by: halil.cakal <halil.cakal@est.tech>

2 files changed, 7 insertions, 53 deletions

diff --git a/src/test/groovy/org/onap/cps/ncmp/dmi/api/kafka/MessagingBaseSpec.groovy b/src/test/groovy/org/onap/cps/ncmp/dmi/api/kafka/MessagingBaseSpec.groovy
index 59e45c20..e5f00f33 100644
--- a/src/test/groovy/org/onap/cps/ncmp/dmi/api/kafka/MessagingBaseSpec.groovy
+++ b/src/test/groovy/org/onap/cps/ncmp/dmi/api/kafka/MessagingBaseSpec.groovy
@@ -20,8 +20,10 @@
 
 package org.onap.cps.ncmp.dmi.api.kafka
 
+import org.apache.kafka.clients.consumer.KafkaConsumer
 import org.apache.kafka.common.serialization.StringDeserializer
 import org.apache.kafka.common.serialization.StringSerializer
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory
 import org.springframework.kafka.core.DefaultKafkaProducerFactory
 import org.springframework.kafka.core.KafkaTemplate
 import org.springframework.kafka.support.serializer.JsonSerializer
@@ -64,6 +66,8 @@ class MessagingBaseSpec extends Specification {
 
     def kafkaTemplate = new KafkaTemplate<>(new DefaultKafkaProducerFactory<Integer, String>(producerConfigProperties()))
 
+    def consumer = new KafkaConsumer<>(consumerConfigProperties('ncmp-group'))
+
     @DynamicPropertySource
     static void registerKafkaProperties(DynamicPropertyRegistry dynamicPropertyRegistry) {
         dynamicPropertyRegistry.add('spring.kafka.bootstrap-servers', kafkaTestContainer::getBootstrapServers)
diff --git a/src/test/groovy/org/onap/cps/ncmp/dmi/notifications/async/AsyncTaskExecutorIntegrationSpec.groovy b/src/test/groovy/org/onap/cps/ncmp/dmi/notifications/async/AsyncTaskExecutorIntegrationSpec.groovy
index b377a37b..96e2c16d 100644
--- a/src/test/groovy/org/onap/cps/ncmp/dmi/notifications/async/AsyncTaskExecutorIntegrationSpec.groovy
+++ b/src/test/groovy/org/onap/cps/ncmp/dmi/notifications/async/AsyncTaskExecutorIntegrationSpec.groovy
@@ -1,6 +1,6 @@
 /*
  * ============LICENSE_START=======================================================
- * Copyright (C) 2022 Nordix Foundation
+ * Copyright (C) 2022-2023 Nordix Foundation
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the 'License');
  * you may not use this file except in compliance with the License.
@@ -21,71 +21,26 @@
 package org.onap.cps.ncmp.dmi.notifications.async
 
 import com.fasterxml.jackson.databind.ObjectMapper
-import org.apache.kafka.clients.consumer.KafkaConsumer
-import org.apache.kafka.common.serialization.StringDeserializer
-import org.apache.kafka.common.serialization.StringSerializer
+import org.onap.cps.ncmp.dmi.api.kafka.MessagingBaseSpec
 import org.onap.cps.ncmp.dmi.exception.HttpClientRequestException
-import org.onap.cps.ncmp.dmi.model.DataAccessRequest
 import org.onap.cps.ncmp.event.model.DmiAsyncRequestResponseEvent
 import org.spockframework.spring.SpringBean
 import org.springframework.boot.test.context.SpringBootTest
 import org.springframework.http.HttpStatus
-import org.springframework.kafka.core.DefaultKafkaProducerFactory
-import org.springframework.kafka.core.KafkaTemplate
-import org.springframework.kafka.support.serializer.JsonSerializer
 import org.springframework.test.annotation.DirtiesContext
-import org.springframework.test.context.DynamicPropertyRegistry
-import org.springframework.test.context.DynamicPropertySource
-import org.testcontainers.containers.KafkaContainer
 import org.testcontainers.spock.Testcontainers
-import org.testcontainers.utility.DockerImageName
-import spock.lang.Specification
 
 import java.time.Duration
 
 @SpringBootTest(classes = [AsyncTaskExecutor, DmiAsyncRequestResponseEventProducer])
 @Testcontainers
 @DirtiesContext
-class AsyncTaskExecutorIntegrationSpec extends Specification {
-
-    static kafkaTestContainer = new KafkaContainer(
-        DockerImageName.parse('confluentinc/cp-kafka:6.2.1')
-    )
-
-    static {
-        Runtime.getRuntime().addShutdownHook(new Thread(kafkaTestContainer::stop))
-    }
-
-    def setupSpec() {
-        kafkaTestContainer.start()
-    }
-
-    def producerConfigProperties = [
-        'bootstrap.servers' : kafkaTestContainer.getBootstrapServers().split(',')[0],
-        'retries'           : 0,
-        'batch.size'        : 16384,
-        'linger.ms'         : 1,
-        'buffer.memory'     : 33554432,
-        'key.serializer'    : StringSerializer,
-        'value.serializer'  : JsonSerializer
-    ]
-
-    def consumerConfigProperties = [
-        'bootstrap.servers' : kafkaTestContainer.getBootstrapServers().split(',')[0],
-        'key.deserializer'  : StringDeserializer,
-        'value.deserializer': StringDeserializer,
-        'auto.offset.reset' : 'earliest',
-        'group.id'          : 'test'
-    ]
-
-    def kafkaTemplate = new KafkaTemplate<>(new DefaultKafkaProducerFactory<Integer, String>(producerConfigProperties))
+class AsyncTaskExecutorIntegrationSpec extends MessagingBaseSpec {
 
     @SpringBean
     DmiAsyncRequestResponseEventProducer cpsAsyncRequestResponseEventProducer =
         new DmiAsyncRequestResponseEventProducer(kafkaTemplate)
 
-    KafkaConsumer<String, Object> consumer = new KafkaConsumer<>(consumerConfigProperties)
-
     def spiedObjectMapper = Spy(ObjectMapper)
 
     def objectUnderTest = new AsyncTaskExecutor(cpsAsyncRequestResponseEventProducer)
@@ -128,9 +83,4 @@ class AsyncTaskExecutorIntegrationSpec extends Specification {
             assert event.getEventContent().getResponseCode() == '500'
     }
 
-    @DynamicPropertySource
-    static void registerKafkaProperties(DynamicPropertyRegistry dynamicPropertyRegistry) {
-        dynamicPropertyRegistry.add('spring.kafka.bootstrap-servers', kafkaTestContainer::getBootstrapServers)
-    }
-
 }
\ No newline at end of file