NCMP : forward bulk response messages to client topic

# Fixing the NullPointer Exception if the 'eventType' header not availabe.
# eventType header is a mandatory header but still we are doing null check for supporting old events (AsyncResponseEvent is not moved to separate kafka headers)

Issue-ID: CPS-1557
Signed-off-by: raviteja.karumuri <raviteja.karumuri@est.tech>
Change-Id: Ie7923d0e2674402fa36cb28cde966575b899cedb

diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/BatchRecordFilterStrategy.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/BatchRecordFilterStrategy.java
index 088e965..2c76599 100644 (file)
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/BatchRecordFilterStrategy.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/BatchRecordFilterStrategy.java
@@ -21,6 +21,7 @@
 package org.onap.cps.ncmp.api.impl.async;
 
 import org.apache.commons.lang3.SerializationUtils;
+import org.apache.kafka.common.header.Header;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
@@ -41,10 +42,14 @@ public class BatchRecordFilterStrategy {
     @Bean
     public RecordFilterStrategy<Object, Object> filterBatchDataResponseEvent() {
         return consumedRecord -> {
-            final String headerValue = SerializationUtils
-                    .deserialize(consumedRecord.headers().lastHeader("eventType").value());
-            return !(headerValue != null
-                    && headerValue.startsWith("org.onap.cps.ncmp.events.async.BatchDataResponseEvent"));
+            final Header eventTypeHeader = consumedRecord.headers().lastHeader("eventType");
+            if (eventTypeHeader != null) {
+                final String eventTypeHeaderValue = SerializationUtils.deserialize(eventTypeHeader.value());
+                return !(eventTypeHeaderValue != null
+                        && eventTypeHeaderValue.startsWith("org.onap.cps.ncmp.events.async.BatchDataResponseEvent"));
+            } else {
+                return true;
+            }
         };
     }
 }