Add DMaaP plugin support

- Update blueprint with DMaaP plugin support.
- Update configuration with new DMaaP plugin format.
- Move Configuration tests to correct package.
- Add latest-staging Docker image tag to pom.
- Change file endings in some files to LF.
- Remove publishing to non authenticated topic.
- Fix some tests which were not executing correctly.
- Fix some Sonar smells.

Issue-ID: DCAEGEN2-1581
Change-Id: I37fbb662419179e3fe9fb8bdf710d6a6e8f0308a
Signed-off-by: JoeOLeary <joseph.o.leary@est.tech>

13 files changed, 324 insertions, 215 deletions

diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/config/ConfigHandler.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/config/ConfigHandler.java
index e50ec6c..e98849e 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/config/ConfigHandler.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/config/ConfigHandler.java
@@ -1,100 +1,102 @@
-/*-

- * ============LICENSE_START=======================================================

- *  Copyright (C) 2019 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.dcaegen2.services.pmmapper.config;

-

-import org.onap.dcaegen2.services.pmmapper.exceptions.CBSConfigException;

-import org.onap.dcaegen2.services.pmmapper.exceptions.CBSServerError;

-import org.onap.dcaegen2.services.pmmapper.exceptions.EnvironmentConfigException;

-import org.onap.dcaegen2.services.pmmapper.exceptions.MapperConfigException;

-import org.onap.dcaegen2.services.pmmapper.model.EnvironmentConfig;

-import org.onap.dcaegen2.services.pmmapper.model.MapperConfig;

-import org.onap.dcaegen2.services.pmmapper.utils.RequestSender;

-import org.onap.dcaegen2.services.pmmapper.utils.RequiredFieldDeserializer;

-

-import org.onap.logging.ref.slf4j.ONAPLogAdapter;

-import org.onap.logging.ref.slf4j.ONAPLogConstants;

-import org.slf4j.LoggerFactory;

-import com.google.gson.GsonBuilder;

-

-/**

- * Handles the retrieval of the component spec-based PM-Mapper Configuration

- * from DCAE.

- */

-

-public class ConfigHandler {

-    private static final ONAPLogAdapter logger = new ONAPLogAdapter(LoggerFactory.getLogger(ConfigHandler.class));

-    private RequestSender sender;

-

-    /**

-     * Creates a ConfigHandler.

-     */

-    public ConfigHandler() {

-        this(new RequestSender());

-    }

-

-    /**

-     * @see ConfigHandler#ConfigHandler()

-     * @param sender A RequestSender

-     */

-    public ConfigHandler(RequestSender sender) {

-        this.sender = sender;

-    }

-

-    /**

-     * Retrieves PM-Mapper Configuration from DCAE's ConfigBinding Service.

-     *

-     * @throws EnvironmentConfigException

-     * @throws ConsulServerError

-     * @throws CBSConfigException

-     * @throws CBSServerError

-     * @throws MapperConfigException

-     */

-    public MapperConfig getMapperConfig() throws EnvironmentConfigException,

-            CBSServerError, MapperConfigException {

-        String mapperConfigJson = "";

-        String cbsSocketAddress = EnvironmentConfig.getCBSHostName() + ":" + EnvironmentConfig.getCBSPort();

-        String requestURL = "http://" + cbsSocketAddress + "/service_component/" + EnvironmentConfig.getServiceName();

-        try {

-            logger.unwrap().info(ONAPLogConstants.Markers.ENTRY, "Fetching pm-mapper configuration from Configbinding Service");

-            mapperConfigJson = sender.send(requestURL);

-        } catch (Exception exception) {

-            throw new CBSServerError("Error connecting to Configbinding Service: ", exception);

-        } finally {

-            logger.unwrap().info(ONAPLogConstants.Markers.EXIT, "Received pm-mapper configuration from ConfigBinding Service:\n{}", mapperConfigJson);

-        }

-

-        return convertMapperConfigToObject(mapperConfigJson);

-    }

-

-    private MapperConfig convertMapperConfigToObject(String mapperConfigJson) throws MapperConfigException {

-        MapperConfig mapperConfig;

-        try {

-            mapperConfig = new GsonBuilder()

-                    .registerTypeAdapter(MapperConfig.class, new RequiredFieldDeserializer<MapperConfig>())

-                    .create()

-                    .fromJson(mapperConfigJson, MapperConfig.class);

-        } catch (Exception exception) {

-            throw new MapperConfigException("Error parsing mapper configuration:\n{}" + mapperConfigJson, exception);

-        }

-

-        logger.unwrap().debug("Mapper configuration:\n{}", mapperConfig);

-        return mapperConfig;

-    }

-}

+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2019 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.dcaegen2.services.pmmapper.config;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import org.onap.dcaegen2.services.pmmapper.exceptions.CBSServerError;
+import org.onap.dcaegen2.services.pmmapper.exceptions.EnvironmentConfigException;
+import org.onap.dcaegen2.services.pmmapper.exceptions.MapperConfigException;
+import org.onap.dcaegen2.services.pmmapper.utils.EnvironmentConfig;
+import org.onap.dcaegen2.services.pmmapper.model.MapperConfig;
+import org.onap.dcaegen2.services.pmmapper.utils.RequestSender;
+import org.onap.dcaegen2.services.pmmapper.utils.RequiredFieldDeserializer;
+
+import org.onap.logging.ref.slf4j.ONAPLogAdapter;
+import org.slf4j.LoggerFactory;
+import com.google.gson.GsonBuilder;
+
+/**
+ * Handles the retrieval of the component spec-based PM-Mapper Configuration
+ * from DCAE.
+ */
+
+public class ConfigHandler {
+    private static final ONAPLogAdapter logger = new ONAPLogAdapter(LoggerFactory.getLogger(ConfigHandler.class));
+    private RequestSender sender;
+    private EnvironmentConfig environmentConfig;
+
+    /**
+     * Creates a ConfigHandler.
+     */
+    public ConfigHandler() {
+        this(new RequestSender(), new EnvironmentConfig());
+    }
+
+    /**
+     * @see ConfigHandler#ConfigHandler()
+     * @param sender A RequestSender
+     */
+    public ConfigHandler(RequestSender sender, EnvironmentConfig environmentConfig) {
+        this.sender = sender;
+        this.environmentConfig = environmentConfig;
+    }
+
+    /**
+     * Retrieves PM-Mapper Configuration from DCAE's ConfigBinding Service.
+     *
+     * @throws EnvironmentConfigException
+     * @throws CBSServerError
+     * @throws MapperConfigException
+     */
+    public MapperConfig getMapperConfig() throws EnvironmentConfigException,
+            CBSServerError, MapperConfigException {
+        String mapperConfigJson = "";
+        String cbsSocketAddress = this.environmentConfig.getCBSHostName() + ":" + this.environmentConfig.getCBSPort();
+        String requestURL = "http://" + cbsSocketAddress + "/service_component/" + this.environmentConfig.getServiceName();
+        try {
+            logger.unwrap().info("Fetching pm-mapper configuration from Configbinding Service");
+            mapperConfigJson = sender.send(requestURL);
+        } catch (Exception exception) {
+            throw new CBSServerError("Error connecting to Configbinding Service: ", exception);
+        } finally {
+            logger.unwrap().info("Received pm-mapper configuration from ConfigBinding Service:\n{}", mapperConfigJson);
+        }
+
+        return convertMapperConfigToObject(mapperConfigJson);
+    }
+
+    private MapperConfig convertMapperConfigToObject(String mapperConfigJson) throws MapperConfigException {
+        MapperConfig mapperConfig;
+        try {
+            JsonObject config = new Gson().fromJson(mapperConfigJson, JsonObject.class);
+            mapperConfig = new GsonBuilder()
+                    .registerTypeAdapter(MapperConfig.class, new RequiredFieldDeserializer<MapperConfig>())
+                    .create()
+                    .fromJson(config, MapperConfig.class);
+        } catch (Exception exception) {
+            String exceptionMessage = "Error parsing configuration, mapper config:\n" + mapperConfigJson;
+            throw new MapperConfigException(exceptionMessage, exception);
+        }
+
+        logger.unwrap().debug("Mapper configuration:\n{}", mapperConfig);
+        return mapperConfig;
+    }
+}
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/config/Configurable.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/config/Configurable.java
index ac2fe57..9431cce 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/config/Configurable.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/config/Configurable.java
@@ -25,6 +25,7 @@ package org.onap.dcaegen2.services.pmmapper.config;
 import org.onap.dcaegen2.services.pmmapper.exceptions.ReconfigurationException;
 import org.onap.dcaegen2.services.pmmapper.model.MapperConfig;
 
+@FunctionalInterface
 public interface Configurable {
 
     void reconfigure(MapperConfig mapperConfig) throws ReconfigurationException;
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfiguration.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfiguration.java
index 37fa8b5..7e2c894 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfiguration.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfiguration.java
@@ -23,6 +23,7 @@ package org.onap.dcaegen2.services.pmmapper.config;
 import io.undertow.server.HttpHandler;
 import io.undertow.server.HttpServerExchange;
 import io.undertow.util.StatusCodes;
+import java.util.List;
 import lombok.Data;
 import org.onap.dcaegen2.services.pmmapper.exceptions.ReconfigurationException;
 import org.onap.dcaegen2.services.pmmapper.model.MapperConfig;
@@ -30,10 +31,8 @@ import org.onap.dcaegen2.services.pmmapper.utils.HttpServerExchangeAdapter;
 import org.onap.logging.ref.slf4j.ONAPLogAdapter;
 import org.slf4j.LoggerFactory;
 
-import java.util.List;
-
 @Data
-public class DynamicConfiguration implements HttpHandler{
+public class DynamicConfiguration implements HttpHandler {
     private static final ONAPLogAdapter logger = new ONAPLogAdapter(LoggerFactory.getLogger(DynamicConfiguration.class));
     private List<Configurable> configurables;
     private MapperConfig originalConfig;
@@ -44,7 +43,7 @@ public class DynamicConfiguration implements HttpHandler{
      * @param configurables list of objects to reconfigure
      * @param originalConfig original config to compare against.
      */
-    public DynamicConfiguration(List<Configurable> configurables, MapperConfig originalConfig){
+    public DynamicConfiguration(List<Configurable> configurables, MapperConfig originalConfig) {
         this.configurables = configurables;
         this.originalConfig = originalConfig;
         this.configHandler = new ConfigHandler();
@@ -79,6 +78,7 @@ public class DynamicConfiguration implements HttpHandler{
                 } catch (ReconfigurationException e) {
                     responseCode = StatusCodes.INTERNAL_SERVER_ERROR;
                     responseMessage = StatusCodes.INTERNAL_SERVER_ERROR_STRING;
+                    logger.unwrap().error("Failed to apply configuration update, reverting to original config", e);
                     applyConfiguration(this.originalConfig);
                 }
             }
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/datarouter/EventReceiver.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/datarouter/EventReceiver.java
index 77c8153..42dba87 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/datarouter/EventReceiver.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/datarouter/EventReceiver.java
@@ -26,6 +26,7 @@ import org.onap.dcaegen2.services.pmmapper.model.Event;
 /**
  * Sink for Events received from the data router subscriber.
  */
+@FunctionalInterface
 public interface EventReceiver {
     void receive(Event event);
 }
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/filtering/MeasFilterHandler.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/filtering/MeasFilterHandler.java
index 2dc4ae9..0438530 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/filtering/MeasFilterHandler.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/filtering/MeasFilterHandler.java
@@ -20,14 +20,9 @@
 
 package org.onap.dcaegen2.services.pmmapper.filtering;
 
-import java.math.BigInteger;
 import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
-import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import org.apache.commons.io.FilenameUtils;
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/filtering/MetadataFilter.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/filtering/MetadataFilter.java
index 1fb6019..d647c40 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/filtering/MetadataFilter.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/filtering/MetadataFilter.java
@@ -99,6 +99,7 @@ public class MetadataFilter {
         return true;
     }
 
+    @FunctionalInterface
     interface Validator<A, B> {
         boolean validate(A filter, B metadata);
     }
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/messagerouter/VESPublisher.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/messagerouter/VESPublisher.java
index 46d40e4..744696a 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/messagerouter/VESPublisher.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/messagerouter/VESPublisher.java
@@ -64,11 +64,10 @@ public class VESPublisher {
         try {

             String topicUrl = config.getPublisherTopicUrl();

             ves = ves.replaceAll("\n", "");

-            String userCredentials =  topicUrl.startsWith("https") ? Base64.getEncoder()

+            String userCredentials =  Base64.getEncoder()

                 .encodeToString((this.config.getPublisherUserName() + ":" +

                     this.config.getPublisherPassword())

-                    .getBytes(StandardCharsets.UTF_8))

-                : "";

+                    .getBytes(StandardCharsets.UTF_8));

             sender.send("POST", topicUrl, ves, userCredentials);

         } catch (Exception e) {

             throw new MRPublisherException(e.getMessage(), e);

diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/model/MapperConfig.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/model/MapperConfig.java
index 0630d53..385a256 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/model/MapperConfig.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/model/MapperConfig.java
@@ -17,20 +17,24 @@
  * SPDX-License-Identifier: Apache-2.0
  * ============LICENSE_END=========================================================
  */
+
 package org.onap.dcaegen2.services.pmmapper.model;
 
-import org.onap.dcaegen2.services.pmmapper.config.Configurable;
-import org.onap.dcaegen2.services.pmmapper.utils.GSONRequired;
+import com.google.gson.annotations.JsonAdapter;
 import com.google.gson.annotations.SerializedName;
-import lombok.Getter;
-import lombok.AccessLevel;
 import lombok.EqualsAndHashCode;
+import lombok.Getter;
 import lombok.NoArgsConstructor;
+import lombok.ToString;
+import org.onap.dcaegen2.services.pmmapper.config.Configurable;
+import org.onap.dcaegen2.services.pmmapper.utils.DMaaPAdapter;
+import org.onap.dcaegen2.services.pmmapper.utils.GSONRequired;
 
 @Getter
 @EqualsAndHashCode
 @NoArgsConstructor
-public class MapperConfig implements Configurable{
+@ToString
+public class MapperConfig implements Configurable {
 
     public static final String CLIENT_NAME = "pm-mapper";
 
@@ -55,16 +59,6 @@ public class MapperConfig implements Configurable{
     private String trustStorePassPath;
 
     @GSONRequired
-    @Getter(AccessLevel.PRIVATE)
-    @SerializedName("streams_subscribes")
-    private StreamsSubscribes streamsSubscribes;
-
-    @GSONRequired
-    @Getter(AccessLevel.PRIVATE)
-    @SerializedName("streams_publishes")
-    private StreamsPublishes streamsPublishes;
-
-    @GSONRequired
     @SerializedName("dmaap_dr_delete_endpoint")
     private String dmaapDRDeleteEndpoint;
 
@@ -72,98 +66,48 @@ public class MapperConfig implements Configurable{
     @SerializedName("pm-mapper-filter")
     private MeasFilterConfig filterConfig;
 
-    public String getSubscriberIdentity(){
-        return this.getStreamsSubscribes().getDmaapSubscriber().getDmaapInfo().getSubscriberId();
-    }
-
-    public String getPublisherTopicUrl() {
-        return this.getStreamsPublishes().getDmaapPublisher().getDmaapInfo().getTopicUrl();
-    }
-
-    public boolean dmaapInfoEquals(MapperConfig mapperConfig){
-        return this
-                .getStreamsSubscribes()
-                .getDmaapSubscriber()
-                .getDmaapInfo()
-                .equals(mapperConfig.getStreamsSubscribes().getDmaapSubscriber().getDmaapInfo());
-    }
-
-    @Getter
-    @EqualsAndHashCode
-    private class StreamsSubscribes {
-        @GSONRequired
-        @SerializedName("dmaap_subscriber")
-        DmaapSubscriber dmaapSubscriber;
-    }
-
-    @Getter
-    @EqualsAndHashCode
-    class DmaapSubscriber {
-        @GSONRequired
-        @SerializedName("dmaap_info")
-        DmaapInfo dmaapInfo;
-    }
+    @GSONRequired
+    @SerializedName("aaf_identity")
+    private String aafUsername;
 
-    @Getter
-    @EqualsAndHashCode
-    private class StreamsPublishes {
-        @GSONRequired
-        @SerializedName("dmaap_publisher")
-        DmaapPublisher dmaapPublisher;
-    }
+    @GSONRequired
+    @SerializedName("aaf_password")
+    private String aafPassword;
 
-    @Getter
-    @EqualsAndHashCode
-    class DmaapPublisher {
-        @GSONRequired
-        @SerializedName("dmaap_info")
-        DmaapInfo dmaapInfo;
+    @GSONRequired
+    @SerializedName("streams_subscribes")
+    @JsonAdapter(DMaaPAdapter.class)
+    private SubscriberConfig subscriberConfig;
 
-        @SerializedName("aaf_username")
-        private String aafUsername;
+    @GSONRequired
+    @SerializedName("streams_publishes")
+    @JsonAdapter(DMaaPAdapter.class)
+    private PublisherConfig publisherConfig;
 
-        @SerializedName("aaf_password")
-        private String aafPassword;
+    public String getSubscriberIdentity() {
+        return this.getSubscriberConfig().getSubscriberId();
     }
 
-    @Getter
-    @EqualsAndHashCode
-    class DmaapInfo {
-        private String location;
-        private String username;
-        private String password;
-
-        @SerializedName("delivery_url")
-        private String deliveryUrl;
-
-        @SerializedName("subscriber_id")
-        private String subscriberId;
-
-        @SerializedName("client_role")
-        private String clientRole;
-
-        @SerializedName("client_id")
-        private String clientId;
-
-        @SerializedName("topic_url")
-        private String topicUrl;
+    public String getPublisherTopicUrl() {
+        return this.getPublisherConfig().getTopicUrl();
     }
 
     public String getPublisherUserName() {
-        return this.getStreamsPublishes().getDmaapPublisher().getAafUsername();
+        return this.getAafUsername();
     }
 
     public String getPublisherPassword() {
-        return this.getStreamsPublishes().getDmaapPublisher().getAafPassword();
+        return this.getAafPassword();
     }
 
-    @Override
     public void reconfigure(MapperConfig mapperConfig) {
-        if(!this.equals(mapperConfig)) {
+        if (!this.equals(mapperConfig)) {
             this.filterConfig = mapperConfig.getFilterConfig();
-            this.streamsSubscribes = mapperConfig.getStreamsSubscribes();
-            this.streamsPublishes = mapperConfig.getStreamsPublishes();
+            this.publisherConfig = mapperConfig.getPublisherConfig();
+            this.subscriberConfig = mapperConfig.getSubscriberConfig();
             this.dmaapDRDeleteEndpoint = mapperConfig.getDmaapDRDeleteEndpoint();
+            this.aafUsername = mapperConfig.getAafUsername();
+            this.aafPassword = mapperConfig.getAafPassword();
         }
     }
 }
\ No newline at end of file
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/model/PublisherConfig.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/model/PublisherConfig.java
new file mode 100644
index 0000000..16ab941
--- /dev/null
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/model/PublisherConfig.java
@@ -0,0 +1,45 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2019 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.dcaegen2.services.pmmapper.model;
+
+import com.google.gson.annotations.SerializedName;
+import lombok.Data;
+import org.onap.dcaegen2.services.pmmapper.utils.GSONRequired;
+
+@Data
+public class PublisherConfig {
+    @GSONRequired
+    @SerializedName("topic_url")
+    private String topicUrl;
+
+    @GSONRequired
+    @SerializedName("client_role")
+    private String clientRole;
+
+    @GSONRequired
+    @SerializedName("client_id")
+    private String clientId;
+
+    @GSONRequired
+    @SerializedName("location")
+    private String clusterLocation;
+
+}
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/model/SubscriberConfig.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/model/SubscriberConfig.java
new file mode 100644
index 0000000..65f680b
--- /dev/null
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/model/SubscriberConfig.java
@@ -0,0 +1,47 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2019 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.dcaegen2.services.pmmapper.model;
+
+import com.google.gson.annotations.SerializedName;
+import lombok.Data;
+import org.onap.dcaegen2.services.pmmapper.utils.GSONRequired;
+
+@Data
+public class SubscriberConfig {
+    @GSONRequired
+    @SerializedName("username")
+    private String username;
+
+    @GSONRequired
+    @SerializedName("password")
+    private String password;
+
+    @GSONRequired
+    @SerializedName("location")
+    private String drLocation;
+
+    @SerializedName("delivery_url")
+    private String deliveryUrl;
+
+    @GSONRequired
+    @SerializedName("subscriber_id")
+    private String subscriberId;
+}
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/DMaaPAdapter.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/DMaaPAdapter.java
new file mode 100644
index 0000000..fa01740
--- /dev/null
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/DMaaPAdapter.java
@@ -0,0 +1,81 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2019 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.dcaegen2.services.pmmapper.utils;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonParseException;
+import com.google.gson.TypeAdapter;
+import com.google.gson.stream.JsonReader;
+import com.google.gson.stream.JsonWriter;
+import java.io.IOException;
+import java.util.Optional;
+import org.onap.dcaegen2.services.pmmapper.model.PublisherConfig;
+import org.onap.dcaegen2.services.pmmapper.model.SubscriberConfig;
+
+public class DMaaPAdapter extends TypeAdapter<Object> {
+    private static final String PUBLISHER = "dmaap_publisher";
+    private static final String SUBSCRIBER = "dmaap_subscriber";
+    private static final String DMAAP_INFO = "dmaap_info";
+
+    @Override
+    public void write(JsonWriter jsonWriter, Object dmaapObj) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public Object read(JsonReader jsonReader) throws IOException {
+        jsonReader.beginObject();
+        String rootName = jsonReader.nextName();
+        Class configClass = getConfigClass(rootName);
+
+        Object generatedConfig = null;
+        jsonReader.beginObject();
+        while (jsonReader.hasNext()) {
+            if (jsonReader.nextName().equals(DMAAP_INFO)) {
+                generatedConfig  = new Gson().fromJson(jsonReader, configClass);
+            } else {
+                jsonReader.skipValue();
+            }
+        }
+        if (generatedConfig == null) {
+            throw new JsonParseException("Failed to Identify DMaaP Object");
+        }
+
+        jsonReader.endObject();
+        jsonReader.endObject();
+        return generatedConfig;
+    }
+    private Class getConfigClass(String rootName) {
+        Class configClass;
+        switch (rootName) {
+            case PUBLISHER:
+                configClass = PublisherConfig.class;
+                break;
+            case SUBSCRIBER:
+                configClass = SubscriberConfig.class;
+                break;
+            default:
+                String reason = String.format("This adapter expects one of: [ %s, %s]", PUBLISHER, SUBSCRIBER);
+                throw new IllegalArgumentException(reason);
+        }
+        return configClass;
+    }
+}
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/model/EnvironmentConfig.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/EnvironmentConfig.java
index e2c1c6e..8457989 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/model/EnvironmentConfig.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/EnvironmentConfig.java
@@ -17,39 +17,32 @@
  * SPDX-License-Identifier: Apache-2.0

  * ============LICENSE_END=========================================================

  */

-package org.onap.dcaegen2.services.pmmapper.model;

+package org.onap.dcaegen2.services.pmmapper.utils;

 

 import java.util.Optional;

 

 import org.onap.dcaegen2.services.pmmapper.exceptions.EnvironmentConfigException;

 

 public class EnvironmentConfig {

-

     public static final int DEFAULT_CBS_PORT = 10000;

     public static final String ENV_CBS_HOST_KEY = "CONFIG_BINDING_SERVICE_SERVICE_HOST";

     public static final String ENV_CBS_PORT_KEY = "CONFIG_BINDING_SERVICE_SERVICE_PORT";

     public static final String ENV_SERVICE_NAME_KEY = "HOSTNAME";

 

-    public static String getServiceName() throws EnvironmentConfigException {

+    public String getServiceName() throws EnvironmentConfigException {

         return Optional.ofNullable(System.getenv(ENV_SERVICE_NAME_KEY))

                 .orElseThrow(() -> new EnvironmentConfigException(

                         ENV_SERVICE_NAME_KEY+ " environment variable must be defined prior to pm-mapper initialization."));

     }

 

-    public static String getCBSHostName() throws EnvironmentConfigException {

+    public String getCBSHostName() throws EnvironmentConfigException {

         return Optional.ofNullable(System.getenv(ENV_CBS_HOST_KEY))

                 .orElseThrow(() -> new EnvironmentConfigException(

                         ENV_CBS_HOST_KEY+ " environment variable must be defined prior to pm-mapper initialization."));

     }

 

-    public static Integer getCBSPort() throws EnvironmentConfigException {

-        Integer port = DEFAULT_CBS_PORT;

-        try {

-            port = Optional.ofNullable(System.getenv(ENV_CBS_PORT_KEY))

+    public Integer getCBSPort() {

+         return Optional.ofNullable(System.getenv(ENV_CBS_PORT_KEY))

                     .map(Integer::valueOf).orElse(DEFAULT_CBS_PORT);

-        } catch (NumberFormatException e) {

-            throw new EnvironmentConfigException(ENV_CBS_PORT_KEY + " must be valid: " + port);

-        }

-        return port;

     }

 }

diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/RequiredFieldDeserializer.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/RequiredFieldDeserializer.java
index 258b831..3520a19 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/RequiredFieldDeserializer.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/RequiredFieldDeserializer.java
@@ -41,7 +41,7 @@ import java.util.stream.Stream;
 public class RequiredFieldDeserializer<T> implements JsonDeserializer<T> {
 
     @Override
-    public T deserialize(JsonElement jsonElement, Type type, JsonDeserializationContext jsonDeserializationContext) throws JsonParseException {
+    public T deserialize(JsonElement jsonElement, Type type, JsonDeserializationContext jsonDeserializationContext) {
         T obj = new Gson().fromJson(jsonElement, type);
         validateRequiredFields(obj.getClass().getDeclaredFields(), obj);
         return obj;