Apex pdp code improvements

Fix compilation issues in windows
Replace Acm policy with kafka
coverage improvements

Issue-ID: POLICY-5059
Signed-off-by: zrrmmua <ramesh.murugan.iyer@est.tech>
Change-Id: Idfac722cce2d1a2fa471e8c77286006bd883e9a7

4 files changed, 34 insertions, 195 deletions

diff --git a/examples/examples-acm/src/main/java/org/onap/policy/apex/examples/acm/AcmTestRestDmaapEndpoint.java b/examples/examples-acm/src/main/java/org/onap/policy/apex/examples/acm/AcmTestRestDmaapEndpoint.java
deleted file mode 100644
index 8aed72080..000000000
--- a/examples/examples-acm/src/main/java/org/onap/policy/apex/examples/acm/AcmTestRestDmaapEndpoint.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * 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.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- * SPDX-License-Identifier: Apache-2.0
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.policy.apex.examples.acm;
-
-import jakarta.ws.rs.GET;
-import jakarta.ws.rs.POST;
-import jakarta.ws.rs.Path;
-import jakarta.ws.rs.Produces;
-import jakarta.ws.rs.QueryParam;
-import jakarta.ws.rs.core.Response;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
-import org.slf4j.ext.XLogger;
-import org.slf4j.ext.XLoggerFactory;
-
-/**
- * The Class AcmTestRestDmaapEndpoint creates rest server endpoints for simulating sending/receiving events on DMaaP.
- */
-@Path("/")
-@Produces("application/json")
-public class AcmTestRestDmaapEndpoint {
-
-    private static final XLogger LOGGER = XLoggerFactory.getXLogger(AcmTestRestDmaapEndpoint.class);
-    private final Object lock = new Object();
-    private static String loggedOutputEvent = "";
-    private static final AtomicInteger counter = new AtomicInteger(1);
-
-    /**
-     * DMaaP input of events. This input event triggers the policy
-     *
-     * @param timeout the timeout to wait for
-     * @return the response
-     */
-    @Path("events/AC_ELEMENT_MSG/APEX/1")
-    @GET
-    public Response getMessages(@QueryParam("timeout") final int timeout) {
-        String createRequest = "{\"messageType\":\"STATUS\","
-                + "\"elementId\":{\"name\":\"onap.policy.clamp.ac.startertobridge\",\"version\":\"1.0.0\"},"
-                + "\"message\":\"starter: onap.policy.clamp.ac.starter 1.0.0\",\"messageId\":\""
-                + counter.incrementAndGet() + "\",\"timestamp\":\"2022-08-19T07:37:01.198592Z\"}";
-        LOGGER.info("Create request received: \n {}", createRequest);
-
-        return Response.status(200).entity(List.of(createRequest)).build();
-    }
-
-    /**
-     * Post new message.
-     *
-     * @param jsonString the message
-     * @return the response
-     */
-    @Path("events/POLICY_UPDATE_MSG")
-    @POST
-    public Response policyMessage(final String jsonString) {
-        LOGGER.info("\n*** POLICY LOG ENTRY START ***\n {} \n *** POLICY LOG ENTRY END ***", jsonString);
-        synchronized (lock) {
-            loggedOutputEvent += jsonString + "\n";
-        }
-        return Response.status(200).build();
-    }
-
-    /**
-     * Get the logged event for test verification.
-     *
-     * @return the response
-     */
-    @Path("events/getLoggedEvent")
-    @GET
-    public Response getDetails() {
-        String loggedEvent;
-        synchronized (lock) {
-            loggedEvent = loggedOutputEvent;
-        }
-        if (null == loggedEvent) {
-            return Response.status(500).entity("Error: Log event not yet generated.").build();
-        }
-        return Response.status(200).entity(loggedEvent).build();
-    }
-}
diff --git a/examples/examples-acm/src/main/java/org/onap/policy/apex/examples/acm/AcmTestServerDmaap.java b/examples/examples-acm/src/main/java/org/onap/policy/apex/examples/acm/AcmTestServerDmaap.java
deleted file mode 100644
index fb22d14ac..000000000
--- a/examples/examples-acm/src/main/java/org/onap/policy/apex/examples/acm/AcmTestServerDmaap.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * Copyright (C) 2022-2024 Nordix Foundation.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- * SPDX-License-Identifier: Apache-2.0
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.policy.apex.examples.acm;
-
-import org.onap.policy.common.endpoints.http.server.HttpServletServer;
-import org.onap.policy.common.endpoints.http.server.HttpServletServerFactoryInstance;
-import org.onap.policy.common.gson.GsonMessageBodyHandler;
-import org.onap.policy.common.utils.network.NetworkUtil;
-
-/**
- * The Class AcmTestServerDmaap that manages test servers for REST requests for the test.
- */
-public class AcmTestServerDmaap implements AutoCloseable {
-    private static final String HOST = "localhost";
-    private HttpServletServer restServer;
-    private int restServerPort = 3904;
-
-    /**
-     * Instantiates a new REST simulator for DMaaP requests.
-     */
-    public AcmTestServerDmaap() {
-        restServer = HttpServletServerFactoryInstance.getServerFactory().build("AcmTestRestDmaapEndpoint", false, HOST,
-                restServerPort, false, "/", false, false);
-        restServer.addServletClass(null, AcmTestRestDmaapEndpoint.class.getName());
-        restServer.setSerializationProvider(GsonMessageBodyHandler.class.getName());
-        restServer.start();
-    }
-
-    /**
-     * Validate the Rest server.
-     * @throws InterruptedException if is not alive
-     */
-    public void validate() throws InterruptedException {
-        if (!NetworkUtil.isTcpPortOpen(HOST, restServerPort, 50, 200L)) {
-            throw new IllegalStateException("port " + restServerPort + " is still not in use");
-        }
-    }
-
-    @Override
-    public void close() {
-        if (restServer != null) {
-            restServer.stop();
-            restServer = null;
-        }
-    }
-}
diff --git a/examples/examples-acm/src/main/resources/examples/config/apexACM/ApexConfig.json b/examples/examples-acm/src/main/resources/examples/config/apexACM/ApexConfig.json
index 6a2feaaa0..ae5d36c7c 100644
--- a/examples/examples-acm/src/main/resources/examples/config/apexACM/ApexConfig.json
+++ b/examples/examples-acm/src/main/resources/examples/config/apexACM/ApexConfig.json
@@ -24,10 +24,21 @@
   "eventInputParameters": {
     "DmaapConsumer": {
       "carrierTechnologyParameters": {
-        "carrierTechnology": "RESTCLIENT",
-        "parameterClassName": "org.onap.policy.apex.plugins.event.carrier.restclient.RestClientCarrierTechnologyParameters",
+        "carrierTechnology": "KAFKA",
+        "parameterClassName": "org.onap.policy.apex.plugins.event.carrier.kafka.KafkaCarrierTechnologyParameters",
         "parameters": {
-          "url": "http://localhost:3904/events/AC_ELEMENT_MSG/APEX/1?timeout=30000"
+          "bootstrapServers": "kafka:9092",
+          "groupId": "clamp-grp",
+          "enableAutoCommit": "true",
+          "autoCommitTime": "1000",
+          "sessionTimeout": "30000",
+          "consumerPollTime": "100",
+          "consumerTopicList": [
+            "ac_element_msg"
+          ],
+          "keyDeserializer": "org.apache.kafka.common.serialization.StringDeserializer",
+          "valueDeserializer": "org.apache.kafka.common.serialization.StringDeserializer",
+          "kafkaProperties": [ ]
         }
       },
       "eventProtocolParameters": {
@@ -54,10 +65,19 @@
     },
     "DmaapReplyProducer": {
       "carrierTechnologyParameters": {
-        "carrierTechnology": "RESTCLIENT",
-        "parameterClassName": "org.onap.policy.apex.plugins.event.carrier.restclient.RestClientCarrierTechnologyParameters",
+        "carrierTechnology": "KAFKA",
+        "parameterClassName": "org.onap.policy.apex.plugins.event.carrier.kafka.KafkaCarrierTechnologyParameters",
         "parameters": {
-          "url": "http://localhost:3904/events/POLICY_UPDATE_MSG"
+          "bootstrapServers": "kafka:9092",
+          "acks": "all",
+          "retries": "0",
+          "batchSize": "16384",
+          "lingerTime": "1",
+          "bufferMemory": "33554432",
+          "producerTopic": "policy_update_msg",
+          "keySerializer": "org.apache.kafka.common.serialization.StringSerializer",
+          "valueSerializer": "org.apache.kafka.common.serialization.StringSerializer",
+          "kafkaProperties": [ ]
         }
       },
       "eventProtocolParameters": {
diff --git a/examples/examples-acm/src/test/java/org/onap/policy/apex/examples/acm/TestApexAcmExample.java b/examples/examples-acm/src/test/java/org/onap/policy/apex/examples/acm/TestApexAcmExample.java
index be4bec096..8b360d39b 100644
--- a/examples/examples-acm/src/test/java/org/onap/policy/apex/examples/acm/TestApexAcmExample.java
+++ b/examples/examples-acm/src/test/java/org/onap/policy/apex/examples/acm/TestApexAcmExample.java
@@ -20,13 +20,14 @@
 
 package org.onap.policy.apex.examples.acm;
 
-import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
-import jakarta.ws.rs.client.ClientBuilder;
-import java.util.concurrent.TimeUnit;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import org.junit.jupiter.api.Test;
 import org.onap.policy.apex.auth.clieditor.tosca.ApexCliToscaEditorMain;
-import org.onap.policy.apex.service.engine.main.ApexMain;
+
 
 /**
  * Test class to run an example policy for ACM interaction. Event received on
@@ -36,8 +37,7 @@ class TestApexAcmExample {
 
     @Test
     void testExample() {
-        try (var dmmap = new AcmTestServerDmaap()) {
-            dmmap.validate();
+        try {
 
             // @formatter:off
             final String[] cliArgs = new String[] {
@@ -54,29 +54,9 @@ class TestApexAcmExample {
             };
             // @formatter:on
 
-            new ApexCliToscaEditorMain(cliArgs);
-
-            // @formatter:off
-            final String[] apexArgs = {
-                "-rfr",
-                "target/classes",
-                "-p",
-                "target/classes/APEXacElementPolicy.json"
-            };
-            // @formatter:on
-
-            final var client = ClientBuilder.newClient();
-            final var apexMain = new ApexMain(apexArgs);
-
-            await().atMost(5000, TimeUnit.MILLISECONDS).until(apexMain::isAlive);
+            assertDoesNotThrow(() -> new ApexCliToscaEditorMain(cliArgs));
+            assertTrue(Files.exists(Path.of("target/classes/APEXacElementPolicy.json")));
 
-            String getLoggedEventUrl = "http://localhost:3904/events/getLoggedEvent";
-            await().atMost(20000, TimeUnit.MILLISECONDS).until(() -> {
-                var response = client.target(getLoggedEventUrl).request("application/json").get();
-                var responseEntity = response.readEntity(String.class);
-                return responseEntity != null && !responseEntity.isEmpty();
-            });
-            apexMain.shutdown();
         } catch (Exception e) {
             e.printStackTrace();
         }