From: Lathish Date: Wed, 29 Jan 2020 12:59:56 +0000 (+0000) Subject: Exception test Fix X-Git-Tag: 1.0.1~15^2 X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=commitdiff_plain;h=refs%2Fchanges%2F69%2F2369%2F3;p=nonrtric.git Exception test Fix Issue-ID: NONRTRIC-107 Change-Id: Ic4f43010b920e338d2c25660565afb25877296fb Signed-off-by: Lathish --- diff --git a/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumer.java b/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumer.java index 889dfafa..c132fa28 100644 --- a/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumer.java +++ b/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumer.java @@ -49,11 +49,6 @@ public interface DmaapMessageConsumer { */ public boolean isAlive(); - /** - * To Stop the DMAAP Listener - */ - public void stopConsumer(); - /** * It's a infinite loop run every configured seconds to fetch the message from DMAAP. This method can be stop by * setting the alive flag to false diff --git a/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumerImpl.java b/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumerImpl.java index c7dab584..bd3ba69f 100644 --- a/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumerImpl.java +++ b/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumerImpl.java @@ -20,8 +20,11 @@ package org.oransc.policyagent.dmaap; +import com.google.common.collect.Iterables; import java.io.IOException; import java.util.Properties; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.onap.dmaap.mr.client.MRClientFactory; import org.onap.dmaap.mr.client.MRConsumer; import org.onap.dmaap.mr.client.response.MRConsumerResponse; @@ -42,16 +45,18 @@ public class DmaapMessageConsumerImpl implements DmaapMessageConsumer { private boolean alive = false; private final ApplicationConfig applicationConfig; protected MRConsumer consumer; - private MRConsumerResponse response = null; @Autowired private DmaapMessageHandler dmaapMessageHandler; + private final long FETCHTIMEOUT = 30000; + + private CountDownLatch sleep = new CountDownLatch(1); @Autowired public DmaapMessageConsumerImpl(ApplicationConfig applicationConfig) { this.applicationConfig = applicationConfig; } - @Scheduled(fixedRate = 1000 * 10) // , initialDelay=60000) + @Scheduled(fixedRate = 1000 * 10) @Override public void run() { if (!alive) { @@ -60,9 +65,11 @@ public class DmaapMessageConsumerImpl implements DmaapMessageConsumer { if (this.alive) { try { Iterable dmaapMsgs = fetchAllMessages(); - logger.debug("Fetched all the messages from DMAAP and will start to process the messages"); - for (String msg : dmaapMsgs) { - processMsg(msg); + if (dmaapMsgs != null && Iterables.size(dmaapMsgs) > 0) { + logger.debug("Fetched all the messages from DMAAP and will start to process the messages"); + for (String msg : dmaapMsgs) { + processMsg(msg); + } } } catch (Exception e) { logger.error("{}: cannot fetch because of ", this, e.getMessage(), e); @@ -71,15 +78,17 @@ public class DmaapMessageConsumerImpl implements DmaapMessageConsumer { } private Iterable fetchAllMessages() { - response = consumer.fetchWithReturnConsumerResponse(); - if (response == null) { + MRConsumerResponse response = null; + try { + response = consumer.fetchWithReturnConsumerResponse(); + } catch (Exception e) { + logger.error("Failed to get message from DMAAP", e); + } + if (response == null || !"200".equals(response.getResponseCode())) { logger.warn("{}: DMaaP NULL response received", this); + sleepAfterFailure(); } else { logger.debug("DMaaP consumer received {} : {}", response.getResponseCode(), response.getResponseMessage()); - if (!"200".equals(response.getResponseCode())) { - logger.error("DMaaP consumer received: {} : {}", response.getResponseCode(), - response.getResponseMessage()); - } } return response.getActualMessages(); } @@ -117,8 +126,12 @@ public class DmaapMessageConsumerImpl implements DmaapMessageConsumer { return alive; } - @Override - public void stopConsumer() { - alive = false; + private void sleepAfterFailure() { + logger.warn("DMAAP message Consumer is put to Sleep for {} milliseconds", FETCHTIMEOUT); + try { + sleep.await(FETCHTIMEOUT, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + logger.error("Failed to put the thread to sleep: {}", e); + } } } diff --git a/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageHandler.java b/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageHandler.java index 172fe989..61c1e1e2 100644 --- a/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageHandler.java +++ b/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageHandler.java @@ -45,6 +45,7 @@ public class DmaapMessageHandler { private static final Logger logger = LoggerFactory.getLogger(DmaapMessageHandler.class); + private boolean initialize = false; @Autowired private ObjectMapper mapper; @Autowired @@ -57,8 +58,9 @@ public class DmaapMessageHandler { // The publish properties is corrupted. It contains the subscribe property values. @Async("threadPoolTaskExecutor") public void handleDmaapMsg(String msg) { - logger.debug("Message ---------->{}", msg); - init(); + if (!initialize) { + init(); + } DmaapRequestMessage dmaapRequestMessage = null; Optional dmaapResponse = null; // Process the message @@ -166,6 +168,7 @@ public class DmaapMessageHandler { topic = dmaapPublisherConfig.getProperty("topic"); logger.debug("Read the topic & Service Name - {} , {}", host, topic); this.restClient = new AsyncRestClient("http://" + host + "/"); // get this value from application config + initialize = true; } }