diff options
15 files changed, 98 insertions, 98 deletions
diff --git a/bpmn/mso-infrastructure-bpmn/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/CreateAndActivatePnfResourceTest.java b/bpmn/mso-infrastructure-bpmn/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/CreateAndActivatePnfResourceTest.java index 0c001b1192..32ec69e5fa 100644 --- a/bpmn/mso-infrastructure-bpmn/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/CreateAndActivatePnfResourceTest.java +++ b/bpmn/mso-infrastructure-bpmn/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/CreateAndActivatePnfResourceTest.java @@ -49,7 +49,7 @@ public class CreateAndActivatePnfResourceTest extends BaseIntegrationTest { private PnfManagementTestImpl pnfManagementTest; @Autowired - private DmaapClientTestImpl dmaapClientTestImpl; + private KafkaClientTestImpl kafkaClientTestImpl; @Before public void setup() { @@ -60,7 +60,7 @@ public class CreateAndActivatePnfResourceTest extends BaseIntegrationTest { } @Test - public void shouldWaitForMessageFromDmaapAndUpdateAaiEntryWhenAaiEntryExists() { + public void shouldWaitForMessageFromKafkaAndUpdateAaiEntryWhenAaiEntryExists() { // given variables.put(PNF_CORRELATION_ID, PnfManagementTestImpl.ID_WITH_ENTRY); ResourceInput ri = getUpdateResInputObj("OLT"); @@ -72,19 +72,19 @@ public class CreateAndActivatePnfResourceTest extends BaseIntegrationTest { // when ProcessInstance instance = runtimeService.startProcessInstanceByKey("CreateAndActivatePnfResource", "businessKey", variables); - assertThat(instance).isWaitingAt("WaitForDmaapPnfReadyNotification").isWaitingFor("WorkflowMessage"); - dmaapClientTestImpl.sendMessage(); + assertThat(instance).isWaitingAt("WaitForKafkaPnfReadyNotification").isWaitingFor("WorkflowMessage"); + kafkaClientTestImpl.sendMessage(); // then assertThat(instance).isEnded().hasPassedInOrder("CreateAndActivatePnf_StartEvent", "CheckInputs", - "CheckAiiForPnfCorrelationId", "DoesAaiContainInfoAboutPnf", "AaiEntryExists", "InformDmaapClient", - "WaitForDmaapPnfReadyNotification", "CreateRelationId", "AaiEntryUpdated"); + "CheckAiiForPnfCorrelationId", "DoesAaiContainInfoAboutPnf", "AaiEntryExists", "InformKafkaClient", + "WaitForKafkaPnfReadyNotification", "CreateRelationId", "AaiEntryUpdated"); Assertions.assertThat(pnfManagementTest.getServiceAndPnfRelationMap()) .containsOnly(MapEntry.entry(SERVICE_INSTANCE_ID, PnfManagementTestImpl.ID_WITH_ENTRY)); } @Test - public void shouldCreateAaiEntryWaitForMessageFromDmaapAndUpdateAaiEntryWhenNoAaiEntryExists() { + public void shouldCreateAaiEntryWaitForMessageFromKafkaAndUpdateAaiEntryWhenNoAaiEntryExists() { // given variables.put(PNF_CORRELATION_ID, PnfManagementTestImpl.ID_WITHOUT_ENTRY); ResourceInput ri = getUpdateResInputObj("OLT"); @@ -96,13 +96,13 @@ public class CreateAndActivatePnfResourceTest extends BaseIntegrationTest { // when ProcessInstance instance = runtimeService.startProcessInstanceByKey("CreateAndActivatePnfResource", "businessKey", variables); - assertThat(instance).isWaitingAt("WaitForDmaapPnfReadyNotification").isWaitingFor("WorkflowMessage"); - dmaapClientTestImpl.sendMessage(); + assertThat(instance).isWaitingAt("WaitForKafkaPnfReadyNotification").isWaitingFor("WorkflowMessage"); + kafkaClientTestImpl.sendMessage(); // then assertThat(instance).isEnded().hasPassedInOrder("CreateAndActivatePnf_StartEvent", "CheckInputs", "CheckAiiForPnfCorrelationId", "DoesAaiContainInfoAboutPnf", "CreatePnfEntryInAai", "AaiEntryExists", - "InformDmaapClient", "WaitForDmaapPnfReadyNotification", "CreateRelationId", "AaiEntryUpdated"); + "InformKafkaClient", "WaitForKafkaPnfReadyNotification", "CreateRelationId", "AaiEntryUpdated"); Assertions.assertThat(pnfManagementTest.getCreated()).containsOnlyKeys(PnfManagementTestImpl.ID_WITHOUT_ENTRY); Assertions.assertThat(pnfManagementTest.getServiceAndPnfRelationMap()) .containsOnly(MapEntry.entry(SERVICE_INSTANCE_ID, PnfManagementTestImpl.ID_WITHOUT_ENTRY)); diff --git a/bpmn/mso-infrastructure-bpmn/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/DmaapClientTestImpl.java b/bpmn/mso-infrastructure-bpmn/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/KafkaClientTestImpl.java index 43fbc59b3d..e4c1717752 100644 --- a/bpmn/mso-infrastructure-bpmn/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/DmaapClientTestImpl.java +++ b/bpmn/mso-infrastructure-bpmn/src/test/java/org/onap/so/bpmn/infrastructure/pnf/delegate/KafkaClientTestImpl.java @@ -20,15 +20,14 @@ package org.onap.so.bpmn.infrastructure.pnf.delegate; -import java.util.Map; import java.util.Objects; -import org.onap.so.bpmn.infrastructure.pnf.dmaap.DmaapClient; +import org.onap.so.bpmn.infrastructure.pnf.kafka.KafkaClient; import org.springframework.context.annotation.Primary; import org.springframework.stereotype.Component; @Component @Primary -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/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/CancelDmaapSubscription.java b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/CancelKafkaSubscription.java index 439591a295..d0a6e3a59f 100644 --- a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/CancelDmaapSubscription.java +++ b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/CancelKafkaSubscription.java @@ -22,23 +22,23 @@ package org.onap.so.bpmn.infrastructure.pnf.delegate; import org.camunda.bpm.engine.delegate.DelegateExecution; import org.camunda.bpm.engine.delegate.JavaDelegate; -import org.onap.so.bpmn.infrastructure.pnf.dmaap.DmaapClient; +import org.onap.so.bpmn.infrastructure.pnf.kafka.KafkaClient; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @Component -public class CancelDmaapSubscription implements JavaDelegate { +public class CancelKafkaSubscription implements JavaDelegate { - private DmaapClient dmaapClient; + private KafkaClient kafkaClient; @Override public void execute(DelegateExecution execution) { String pnfCorrelationId = (String) execution.getVariable(ExecutionVariableNames.PNF_CORRELATION_ID); - dmaapClient.unregister(pnfCorrelationId); + kafkaClient.unregister(pnfCorrelationId); } @Autowired - public void setDmaapClient(DmaapClient dmaapClient) { - this.dmaapClient = dmaapClient; + public void setKafkaClient(KafkaClient kafkaClient) { + this.kafkaClient = kafkaClient; } } diff --git a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/InformDmaapClient.java b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/InformKafkaClient.java index 5cbd530a93..6506450c52 100644 --- a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/InformDmaapClient.java +++ b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/InformKafkaClient.java @@ -23,25 +23,25 @@ package org.onap.so.bpmn.infrastructure.pnf.delegate; import org.camunda.bpm.engine.RuntimeService; import org.camunda.bpm.engine.delegate.DelegateExecution; import org.camunda.bpm.engine.delegate.JavaDelegate; -import org.onap.so.bpmn.infrastructure.pnf.dmaap.DmaapClient; +import org.onap.so.bpmn.infrastructure.pnf.kafka.KafkaClient; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @Component -public class InformDmaapClient implements JavaDelegate { +public class InformKafkaClient implements JavaDelegate { - private DmaapClient dmaapClient; + private KafkaClient kafkaClient; @Override public void execute(DelegateExecution execution) { String pnfCorrelationId = (String) execution.getVariable(ExecutionVariableNames.PNF_CORRELATION_ID); RuntimeService runtimeService = execution.getProcessEngineServices().getRuntimeService(); - dmaapClient.registerForUpdate(pnfCorrelationId, () -> runtimeService.createMessageCorrelation("WorkflowMessage") + kafkaClient.registerForUpdate(pnfCorrelationId, () -> runtimeService.createMessageCorrelation("WorkflowMessage") .processInstanceBusinessKey(execution.getProcessBusinessKey()).correlateWithResult()); } @Autowired - public void setDmaapClient(DmaapClient dmaapClient) { - this.dmaapClient = dmaapClient; + public void setKafkaClient(KafkaClient kafkaClient) { + this.kafkaClient = kafkaClient; } } diff --git a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/RegisterForPnfReadyEvent.java b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/RegisterForPnfReadyEvent.java index 9e1b5d5a0b..ef5da92e78 100644 --- a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/RegisterForPnfReadyEvent.java +++ b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/RegisterForPnfReadyEvent.java @@ -26,7 +26,7 @@ import org.camunda.bpm.engine.RuntimeService; import org.camunda.bpm.engine.delegate.DelegateExecution; import org.camunda.bpm.engine.delegate.JavaDelegate; import org.onap.so.bpmn.common.BuildingBlockExecution; -import org.onap.so.bpmn.infrastructure.pnf.dmaap.DmaapClient; +import org.onap.so.bpmn.infrastructure.pnf.kafka.KafkaClient; import org.onap.so.bpmn.servicedecomposition.bbobjects.Pnf; import org.onap.so.bpmn.servicedecomposition.entities.ResourceKey; import org.onap.so.bpmn.servicedecomposition.tasks.ExtractPojosForBB; @@ -45,19 +45,19 @@ import org.springframework.stereotype.Component; public class RegisterForPnfReadyEvent implements JavaDelegate { private static final String ERROR_MESSAGE_PNF_NOT_FOUND = - "pnf resource not found in buildingBlockExecution while registering to dmaap listener"; + "pnf resource not found in buildingBlockExecution while registering to kafka listener"; private static final Logger LOGGER = LoggerFactory.getLogger(RegisterForPnfReadyEvent.class); - private DmaapClient dmaapClient; + private KafkaClient kafkaClient; private ExtractPojosForBB extractPojosForBB; private ExceptionBuilder exceptionBuilder; private String pnfEntryNotificationTimeout; @Autowired - public RegisterForPnfReadyEvent(DmaapClient dmaapClient, ExtractPojosForBB extractPojosForBB, + public RegisterForPnfReadyEvent(KafkaClient kafkaClient, ExtractPojosForBB extractPojosForBB, ExceptionBuilder exceptionBuilder, @Value("${aai.pnfEntryNotificationTimeout}") String pnfEntryNotificationTimeout) { - this.dmaapClient = dmaapClient; + this.kafkaClient = kafkaClient; this.extractPojosForBB = extractPojosForBB; this.exceptionBuilder = exceptionBuilder; this.pnfEntryNotificationTimeout = pnfEntryNotificationTimeout; @@ -69,7 +69,7 @@ public class RegisterForPnfReadyEvent implements JavaDelegate { String pnfName = getPnfName(execution); fillExecution(execution, pnfName); RuntimeService runtimeService = execution.getProcessEngineServices().getRuntimeService(); - dmaapClient.registerForUpdate(pnfName, () -> runtimeService.createMessageCorrelation("WorkflowMessage") + kafkaClient.registerForUpdate(pnfName, () -> runtimeService.createMessageCorrelation("WorkflowMessage") .processInstanceId(execution.getProcessInstanceId()).correlateWithResult()); } catch (BBObjectNotFoundException e) { LOGGER.error(ERROR_MESSAGE_PNF_NOT_FOUND); diff --git a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/dmaap/JsonUtilForPnfCorrelationId.java b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/kafka/JsonUtilForPnfCorrelationId.java index 9cb566f49b..3c65cbaad3 100644 --- a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/dmaap/JsonUtilForPnfCorrelationId.java +++ b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/kafka/JsonUtilForPnfCorrelationId.java @@ -21,7 +21,7 @@ * ============LICENSE_END========================================================= */ -package org.onap.so.bpmn.infrastructure.pnf.dmaap; +package org.onap.so.bpmn.infrastructure.pnf.kafka; import com.google.gson.JsonArray; import com.google.gson.JsonElement; diff --git a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/dmaap/DmaapClient.java b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/kafka/KafkaClient.java index fd7eb153b6..941c565bca 100644 --- a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/dmaap/DmaapClient.java +++ b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/kafka/KafkaClient.java @@ -19,9 +19,9 @@ * ============LICENSE_END========================================================= */ -package org.onap.so.bpmn.infrastructure.pnf.dmaap; +package org.onap.so.bpmn.infrastructure.pnf.kafka; -public interface DmaapClient { +public interface KafkaClient { void registerForUpdate(String pnfCorrelationId, Runnable informConsumer); diff --git a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/dmaap/PnfEventReadyDmaapClient.java b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/kafka/PnfEventReadyKafkaClient.java index 44b16dad28..0d3e0e0230 100644 --- a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/dmaap/PnfEventReadyDmaapClient.java +++ b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/kafka/PnfEventReadyKafkaClient.java @@ -19,7 +19,7 @@ * limitations under the License. * ============LICENSE_END========================================================= */ -package org.onap.so.bpmn.infrastructure.pnf.dmaap; +package org.onap.so.bpmn.infrastructure.pnf.kafka; import java.io.IOException; import java.util.Collections; @@ -36,12 +36,12 @@ import org.springframework.core.env.Environment; import org.springframework.stereotype.Component; @Component -public class PnfEventReadyDmaapClient implements DmaapClient { - private static final Logger logger = LoggerFactory.getLogger(PnfEventReadyDmaapClient.class); +public class PnfEventReadyKafkaClient implements KafkaClient { + private static final Logger logger = LoggerFactory.getLogger(PnfEventReadyKafkaClient.class); private Map<String, Runnable> pnfCorrelationIdToThreadMap; private int topicListenerDelayInSeconds; private volatile ScheduledThreadPoolExecutor executor; - private volatile boolean dmaapThreadListenerIsRunning; + private volatile boolean kafkaThreadListenerIsRunning; private KafkaConsumerImpl consumerForPnfReady; private KafkaConsumerImpl consumerForPnfUpdate; private String pnfReadyTopic; @@ -53,9 +53,9 @@ public class PnfEventReadyDmaapClient implements DmaapClient { @Autowired - public PnfEventReadyDmaapClient(Environment env) throws IOException { + public PnfEventReadyKafkaClient(Environment env) throws IOException { pnfCorrelationIdToThreadMap = new ConcurrentHashMap<>(); - topicListenerDelayInSeconds = env.getProperty("pnf.dmaap.topicListenerDelayInSeconds", Integer.class); + topicListenerDelayInSeconds = env.getProperty("pnf.kafka.topicListenerDelayInSeconds", Integer.class); executor = null; try { consumerForPnfReady = new KafkaConsumerImpl(env.getProperty("pnf.kafka.kafkaBootstrapServers")); @@ -75,8 +75,8 @@ public class PnfEventReadyDmaapClient implements DmaapClient { public synchronized void registerForUpdate(String pnfCorrelationId, Runnable informConsumer) { logger.debug("registering for pnf ready kafka event for pnf correlation id: {}", pnfCorrelationId); pnfCorrelationIdToThreadMap.put(pnfCorrelationId, informConsumer); - if (!dmaapThreadListenerIsRunning) { - startDmaapThreadListener(); + if (!kafkaThreadListenerIsRunning) { + startKafkaThreadListener(); } } @@ -87,31 +87,31 @@ public class PnfEventReadyDmaapClient implements DmaapClient { if (pnfCorrelationIdToThreadMap.isEmpty()) { consumerForPnfUpdate.close(); consumerForPnfReady.close(); - stopDmaapThreadListener(); + stopKafkaThreadListener(); } return runnable; } - private synchronized void startDmaapThreadListener() { - if (!dmaapThreadListenerIsRunning) { + private synchronized void startKafkaThreadListener() { + if (!kafkaThreadListenerIsRunning) { executor = new ScheduledThreadPoolExecutor(1); executor.setContinueExistingPeriodicTasksAfterShutdownPolicy(false); executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); - executor.scheduleWithFixedDelay(new DmaapTopicListenerThread(), 0, topicListenerDelayInSeconds, + executor.scheduleWithFixedDelay(new KafkaTopicListenerThread(), 0, topicListenerDelayInSeconds, TimeUnit.SECONDS); - dmaapThreadListenerIsRunning = true; + kafkaThreadListenerIsRunning = true; } } - private synchronized void stopDmaapThreadListener() { - if (dmaapThreadListenerIsRunning) { + private synchronized void stopKafkaThreadListener() { + if (kafkaThreadListenerIsRunning) { executor.shutdown(); - dmaapThreadListenerIsRunning = false; + kafkaThreadListenerIsRunning = false; executor = null; } } - class DmaapTopicListenerThread implements Runnable { + class KafkaTopicListenerThread implements Runnable { @Override public void run() { try { @@ -141,7 +141,7 @@ public class PnfEventReadyDmaapClient implements DmaapClient { private void informAboutPnfReadyIfPnfCorrelationIdFound(String pnfCorrelationId) { Runnable runnable = unregister(pnfCorrelationId); if (runnable != null) { - logger.debug("dmaap listener gets pnf ready event for pnfCorrelationId: {}", pnfCorrelationId); + logger.debug("kafka listener gets pnf ready event for pnfCorrelationId: {}", pnfCorrelationId); runnable.run(); } } 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); diff --git a/bpmn/so-bpmn-tasks/src/main/java/org/onap/so/bpmn/infrastructure/workflow/tasks/ebb/loader/PnfEBBLoader.java b/bpmn/so-bpmn-tasks/src/main/java/org/onap/so/bpmn/infrastructure/workflow/tasks/ebb/loader/PnfEBBLoader.java index 761219c6be..a44093298d 100644 --- a/bpmn/so-bpmn-tasks/src/main/java/org/onap/so/bpmn/infrastructure/workflow/tasks/ebb/loader/PnfEBBLoader.java +++ b/bpmn/so-bpmn-tasks/src/main/java/org/onap/so/bpmn/infrastructure/workflow/tasks/ebb/loader/PnfEBBLoader.java @@ -82,6 +82,7 @@ public class PnfEBBLoader { aaiResourceIds.add(new Pair<>(WorkflowType.PNF, pnf.getPnfId())); Resource pnfResource = new Resource(WorkflowType.PNF, pnf.getPnfId(), false, serviceResource); org.onap.aai.domain.yang.Pnf aaiPnf = bbInputSetupUtils.getAAIPnf(pnf.getPnfName()); + pnfResource.setInstanceName(pnf.getPnfName()); pnfResource.setModelCustomizationId(aaiPnf.getModelCustomizationId()); pnfResource.setModelVersionId(aaiPnf.getModelVersionId()); resourceList.add(pnfResource); |