[SO] Code improvement in bpmn-infra supporting kafka change1.13.0

Code improvement in bpmn-infra supporting kafka change

Issue-ID: SO-4122
Change-Id: I3924418d16f8f6d9270278f1894e224a216d1cf2
Signed-off-by: sushant53 <sushant.jadhav@t-systems.com>

6 files changed, 46 insertions, 46 deletions

diff --git a/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/CancelDmaapSubscriptionTest.java b/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/CancelKafkaSubscriptionTest.java
index c2e87d57bf..6230dab3fa 100644
--- a/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/CancelDmaapSubscriptionTest.java
+++ b/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/CancelKafkaSubscriptionTest.java
@@ -27,25 +27,25 @@ import static org.mockito.Mockito.when;
 import org.camunda.bpm.engine.delegate.DelegateExecution;
 import org.junit.Test;
 
-public class CancelDmaapSubscriptionTest {
+public class CancelKafkaSubscriptionTest {
 
     private static final String TEST_PNF_CORRELATION_ID = "testPnfCorrelationId";
 
     @Test
     public void shouldCancelSubscription() {
         // given
-        CancelDmaapSubscription delegate = new CancelDmaapSubscription();
-        DmaapClientTestImpl dmaapClientTest = new DmaapClientTestImpl();
-        delegate.setDmaapClient(dmaapClientTest);
+        CancelKafkaSubscription delegate = new CancelKafkaSubscription();
+        KafkaClientTestImpl kafkaClientTest = new KafkaClientTestImpl();
+        delegate.setKafkaClient(kafkaClientTest);
         DelegateExecution delegateExecution = mock(DelegateExecution.class);
         when(delegateExecution.getVariable(eq(ExecutionVariableNames.PNF_CORRELATION_ID)))
                 .thenReturn(TEST_PNF_CORRELATION_ID);
         when(delegateExecution.getProcessBusinessKey()).thenReturn("testBusinessKey");
-        dmaapClientTest.registerForUpdate("testPnfCorrelationId", () -> {
+        kafkaClientTest.registerForUpdate("testPnfCorrelationId", () -> {
         });
         // when
         delegate.execute(delegateExecution);
         // then
-        assertThat(dmaapClientTest.haveRegisteredConsumer()).isFalse();
+        assertThat(kafkaClientTest.haveRegisteredConsumer()).isFalse();
     }
 }
diff --git a/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/InformDmaapClientTest.java b/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/InformKafkaClientTest.java
index 94aa1427a4..d8a102f2b4 100644
--- a/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/InformDmaapClientTest.java
+++ b/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/InformKafkaClientTest.java
@@ -35,40 +35,40 @@ import org.junit.Before;
 import org.junit.Test;
 import org.mockito.InOrder;
 
-public class InformDmaapClientTest {
+public class InformKafkaClientTest {
     @Before
     public void setUp() {
-        informDmaapClient = new InformDmaapClient();
-        dmaapClientTest = new DmaapClientTestImpl();
-        informDmaapClient.setDmaapClient(dmaapClientTest);
+        informKafkaClient = new InformKafkaClient();
+        kafkaClientTest = new KafkaClientTestImpl();
+        informKafkaClient.setKafkaClient(kafkaClientTest);
         delegateExecution = mockDelegateExecution();
     }
 
-    private InformDmaapClient informDmaapClient;
+    private InformKafkaClient informKafkaClient;
 
-    private DmaapClientTestImpl dmaapClientTest;
+    private KafkaClientTestImpl kafkaClientTest;
 
     private DelegateExecution delegateExecution;
 
     private MessageCorrelationBuilder messageCorrelationBuilder;
 
     @Test
-    public void shouldSendListenerToDmaapClient() {
+    public void shouldSendListenerToKafkaClient() {
         // when
-        informDmaapClient.execute(delegateExecution);
+        informKafkaClient.execute(delegateExecution);
         // then
-        assertThat(dmaapClientTest.getPnfCorrelationId()).isEqualTo("testPnfCorrelationId");
-        assertThat(dmaapClientTest.getInformConsumer()).isNotNull();
+        assertThat(kafkaClientTest.getPnfCorrelationId()).isEqualTo("testPnfCorrelationId");
+        assertThat(kafkaClientTest.getInformConsumer()).isNotNull();
         verifyZeroInteractions(messageCorrelationBuilder);
     }
 
     @Test
-    public void shouldSendListenerToDmaapClientAndSendMessageToCamunda() {
+    public void shouldSendListenerToKafkaClientAndSendMessageToCamunda() {
         // when
-        informDmaapClient.execute(delegateExecution);
-        dmaapClientTest.getInformConsumer().run();
+        informKafkaClient.execute(delegateExecution);
+        kafkaClientTest.getInformConsumer().run();
         // then
-        assertThat(dmaapClientTest.getPnfCorrelationId()).isEqualTo("testPnfCorrelationId");
+        assertThat(kafkaClientTest.getPnfCorrelationId()).isEqualTo("testPnfCorrelationId");
         InOrder inOrder = inOrder(messageCorrelationBuilder);
         inOrder.verify(messageCorrelationBuilder).processInstanceBusinessKey("testBusinessKey");
         inOrder.verify(messageCorrelationBuilder).correlateWithResult();
diff --git a/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/DmaapClientTestImpl.java b/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/KafkaClientTestImpl.java
index 0ec0ac8214..c374ba40c2 100644
--- a/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/DmaapClientTestImpl.java
+++ b/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/KafkaClientTestImpl.java
@@ -22,9 +22,9 @@
 package org.onap.so.bpmn.infrastructure.pnf.delegate;
 
 import java.util.Objects;
-import org.onap.so.bpmn.infrastructure.pnf.dmaap.DmaapClient;
+import org.onap.so.bpmn.infrastructure.pnf.kafka.KafkaClient;
 
-public class DmaapClientTestImpl implements DmaapClient {
+public class KafkaClientTestImpl implements KafkaClient {
 
     private String pnfCorrelationId;
     private Runnable informConsumer;
diff --git a/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/RegisterForPnfReadyEventTest.java b/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/RegisterForPnfReadyEventTest.java
index 7ec05fda04..6422e5f88e 100644
--- a/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/RegisterForPnfReadyEventTest.java
+++ b/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/RegisterForPnfReadyEventTest.java
@@ -29,7 +29,7 @@ public class RegisterForPnfReadyEventTest {
 
     private DelegateExecution delegateExecution;
     private ExtractPojosForBB extractPojosForBBMock;
-    private DmaapClientTestImpl dmaapClientTest;
+    private KafkaClientTestImpl kafkaClientTest;
     private MessageCorrelationBuilder messageCorrelationBuilder;
     private ExceptionBuilder exceptionBuilderMock;
     private BuildingBlockExecution buildingBlockExecution;
@@ -39,7 +39,7 @@ public class RegisterForPnfReadyEventTest {
     @Before
     public void init() {
         delegateExecution = prepareExecution();
-        dmaapClientTest = new DmaapClientTestImpl();
+        kafkaClientTest = new KafkaClientTestImpl();
         exceptionBuilderMock = mock(ExceptionBuilder.class);
         extractPojosForBBMock = mock(ExtractPojosForBB.class);
         buildingBlockExecution = new DelegateExecutionImpl(new HashMap<>());
@@ -47,9 +47,9 @@ public class RegisterForPnfReadyEventTest {
     }
 
     @Test
-    public void shouldRegisterForDmaapClient() throws BBObjectNotFoundException {
+    public void shouldRegisterForKafkaClient() throws BBObjectNotFoundException {
         // given
-        testedObject = new RegisterForPnfReadyEvent(dmaapClientTest, extractPojosForBBMock, exceptionBuilderMock,
+        testedObject = new RegisterForPnfReadyEvent(kafkaClientTest, extractPojosForBBMock, exceptionBuilderMock,
                 PNF_ENTRY_NOTIFICATION_TIMEOUT);
         Pnf pnf = new Pnf();
         pnf.setPnfName(PNF_NAME);
@@ -60,13 +60,13 @@ public class RegisterForPnfReadyEventTest {
         verify(delegateExecution).setVariable(ExecutionVariableNames.PNF_CORRELATION_ID, PNF_NAME);
         verify(delegateExecution).setVariable(ExecutionVariableNames.TIMEOUT_FOR_NOTIFICATION,
                 PNF_ENTRY_NOTIFICATION_TIMEOUT);
-        checkIfInformConsumerThreadIsRunProperly(dmaapClientTest);
+        checkIfInformConsumerThreadIsRunProperly(kafkaClientTest);
     }
 
     @Test
     public void pnfNotFoundInBBexecution_WorkflowExIsThrown() throws BBObjectNotFoundException {
         // given
-        testedObject = new RegisterForPnfReadyEvent(dmaapClientTest, extractPojosForBBMock, exceptionBuilderMock,
+        testedObject = new RegisterForPnfReadyEvent(kafkaClientTest, extractPojosForBBMock, exceptionBuilderMock,
                 PNF_ENTRY_NOTIFICATION_TIMEOUT);
         when(extractPojosForBBMock.extractByKey(buildingBlockExecution, ResourceKey.PNF))
                 .thenThrow(BBObjectNotFoundException.class);
@@ -74,13 +74,13 @@ public class RegisterForPnfReadyEventTest {
         testedObject.execute(delegateExecution);
         // then
         verify(exceptionBuilderMock).buildAndThrowWorkflowException(delegateExecution, 7000,
-                "pnf resource not found in buildingBlockExecution while registering to dmaap listener");
+                "pnf resource not found in buildingBlockExecution while registering to kafka listener");
     }
 
     @Test
     public void pnfNameIsNull_WorkflowExIsThrown() throws BBObjectNotFoundException {
         // given
-        testedObject = new RegisterForPnfReadyEvent(dmaapClientTest, extractPojosForBBMock, exceptionBuilderMock,
+        testedObject = new RegisterForPnfReadyEvent(kafkaClientTest, extractPojosForBBMock, exceptionBuilderMock,
                 PNF_ENTRY_NOTIFICATION_TIMEOUT);
         when(extractPojosForBBMock.extractByKey(buildingBlockExecution, ResourceKey.PNF)).thenReturn(new Pnf());
         // when
@@ -92,7 +92,7 @@ public class RegisterForPnfReadyEventTest {
     @Test
     public void pnfEventNotificationTimeoutNotSet_WorkflowExIsThrown() throws BBObjectNotFoundException {
         // given
-        testedObject = new RegisterForPnfReadyEvent(dmaapClientTest, extractPojosForBBMock, exceptionBuilderMock, null);
+        testedObject = new RegisterForPnfReadyEvent(kafkaClientTest, extractPojosForBBMock, exceptionBuilderMock, null);
         when(extractPojosForBBMock.extractByKey(buildingBlockExecution, ResourceKey.PNF)).thenReturn(new Pnf());
         // when
         testedObject.execute(delegateExecution);
@@ -101,8 +101,8 @@ public class RegisterForPnfReadyEventTest {
                 "pnfEntryNotificationTimeout value not defined");
     }
 
-    private void checkIfInformConsumerThreadIsRunProperly(DmaapClientTestImpl dmaapClientTest) {
-        dmaapClientTest.getInformConsumer().run();
+    private void checkIfInformConsumerThreadIsRunProperly(KafkaClientTestImpl kafkaClientTest) {
+        kafkaClientTest.getInformConsumer().run();
         InOrder inOrder = inOrder(messageCorrelationBuilder);
         inOrder.verify(messageCorrelationBuilder).processInstanceId(PROCESS_INSTANCE_ID);
         inOrder.verify(messageCorrelationBuilder).correlateWithResult();
diff --git a/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/dmaap/JsonUtilForPnfCorrelationIdTest.java b/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/kafka/JsonUtilForPnfCorrelationIdTest.java
index f9e4cb4c88..71d60305e4 100644
--- a/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/dmaap/JsonUtilForPnfCorrelationIdTest.java
+++ b/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/kafka/JsonUtilForPnfCorrelationIdTest.java
@@ -20,7 +20,7 @@
  * ============LICENSE_END=========================================================
  */
 
-package org.onap.so.bpmn.infrastructure.pnf.dmaap;
+package org.onap.so.bpmn.infrastructure.pnf.kafka;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import java.util.ArrayList;
diff --git a/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/dmaap/PnfEventReadyDmaapClientTest.java b/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/kafka/PnfEventReadyKafkaClientTest.java
index 546e644fbd..41887e3586 100644
--- a/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/dmaap/PnfEventReadyDmaapClientTest.java
+++ b/bpmn/so-bpmn-infrastructure-common/src/test/java/org/onap/so/bpmn/infrastructure/pnf/kafka/PnfEventReadyKafkaClientTest.java
@@ -20,7 +20,7 @@
  * ============LICENSE_END=========================================================
  */
 
-package org.onap.so.bpmn.infrastructure.pnf.dmaap;
+package org.onap.so.bpmn.infrastructure.pnf.kafka;
 
 
 import static org.junit.Assert.assertEquals;
@@ -42,13 +42,13 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
-import org.onap.so.bpmn.infrastructure.pnf.dmaap.PnfEventReadyDmaapClient.DmaapTopicListenerThread;
+import org.onap.so.bpmn.infrastructure.pnf.kafka.PnfEventReadyKafkaClient.KafkaTopicListenerThread;
 import org.onap.so.client.kafka.KafkaConsumerImpl;
 import org.springframework.core.env.Environment;
 
 
 @RunWith(MockitoJUnitRunner.class)
-public class PnfEventReadyDmaapClientTest {
+public class PnfEventReadyKafkaClientTest {
     private static final String KAFKA_BOOTSTRAP_SERVERS = "localhost:9092";
     private static final String PNF_CORRELATION_ID = "corrTestId";
     private static final String PNF_CORRELATION_ID_NOT_FOUND_IN_MAP = "otherCorrId";
@@ -66,9 +66,9 @@ public class PnfEventReadyDmaapClientTest {
 
     @Mock
     private Environment env;
-    private PnfEventReadyDmaapClient testedObject;
+    private PnfEventReadyKafkaClient testedObject;
 
-    private DmaapTopicListenerThread testedObjectInnerClassThread;
+    private KafkaTopicListenerThread testedObjectInnerClassThread;
     private KafkaConsumerImpl kafkaConsumerMock;
     private Runnable threadMockToNotifyCamundaFlow;
     private ScheduledThreadPoolExecutor executorMock;
@@ -81,10 +81,10 @@ public class PnfEventReadyDmaapClientTest {
         when(env.getProperty(eq("pnf.kafka.consumerId"))).thenReturn(CONSUMER_ID);
         when(env.getProperty(eq("pnf.kafka.consumerIdUpdate"))).thenReturn(CONSUMER_ID_UPDATE);
         when(env.getProperty(eq("pnf.kafka.consumerGroup"))).thenReturn(CONSUMER_GROUP);
-        when(env.getProperty(eq("pnf.dmaap.topicListenerDelayInSeconds"), eq(Integer.class)))
+        when(env.getProperty(eq("pnf.kafka.topicListenerDelayInSeconds"), eq(Integer.class)))
                 .thenReturn(TOPIC_LISTENER_DELAY_IN_SECONDS);
-        testedObject = new PnfEventReadyDmaapClient(env);
-        testedObjectInnerClassThread = testedObject.new DmaapTopicListenerThread();
+        testedObject = new PnfEventReadyKafkaClient(env);
+        testedObjectInnerClassThread = testedObject.new KafkaTopicListenerThread();
         kafkaConsumerMock = mock(KafkaConsumerImpl.class);
         threadMockToNotifyCamundaFlow = mock(Runnable.class);
         executorMock = mock(ScheduledThreadPoolExecutor.class);
@@ -94,7 +94,7 @@ public class PnfEventReadyDmaapClientTest {
     /**
      * Test run method, where the are following conditions:
      * <p>
-     * - DmaapThreadListener is running, flag is set to true
+     * - KafkaThreadListener is running, flag is set to true
      * <p>
      * - map is filled with one entry with the key that we get from response
      * <p>
@@ -130,7 +130,7 @@ public class PnfEventReadyDmaapClientTest {
     /**
      * Test run method, where the are following conditions:
      * <p>
-     * - DmaapThreadListener is running, flag is set to true
+     * - KafkaThreadListener is running, flag is set to true
      * <p>
      * - map is filled with one entry with the pnfCorrelationId that does not match to pnfCorrelationId taken from http
      * response. run method should not do anything with the map not run any thread to notify camunda process
@@ -147,7 +147,7 @@ public class PnfEventReadyDmaapClientTest {
     /**
      * Test run method, where the are following conditions:
      * <p>
-     * - DmaapThreadListener is running, flag is set to true
+     * - KafkaThreadListener is running, flag is set to true
      * <p>
      * - map is filled with one entry with the pnfCorrelationId but no correlation id is taken from HttpResponse run
      * method should not do anything with the map and not run any thread to notify camunda process
@@ -182,7 +182,7 @@ public class PnfEventReadyDmaapClientTest {
         pnfCorrelationToThreadMap.put(PNF_CORRELATION_ID, threadMockToNotifyCamundaFlow);
         pnfCorrelationToThreadMapField.set(testedObject, pnfCorrelationToThreadMap);
 
-        Field threadRunFlag = testedObject.getClass().getDeclaredField("dmaapThreadListenerIsRunning");
+        Field threadRunFlag = testedObject.getClass().getDeclaredField("kafkaThreadListenerIsRunning");
         threadRunFlag.setAccessible(true);
         threadRunFlag.set(testedObject, true);
         threadRunFlag.setAccessible(false);