add test cases after the kafka 11 upgrade changes

Issue-ID: DMAAP-527
Change-Id: I5accb52b74bdb504fdcda0030192dd28fe72ace4
Signed-off-by: sunil unnava <su622b@att.com>

27 files changed, 1309 insertions, 715 deletions

diff --git a/src/main/java/com/att/nsa/dmaap/DMaaPCambriaExceptionMapper.java b/src/main/java/com/att/nsa/dmaap/DMaaPCambriaExceptionMapper.java
index 53c3bed..e5fe8c4 100644
--- a/src/main/java/com/att/nsa/dmaap/DMaaPCambriaExceptionMapper.java
+++ b/src/main/java/com/att/nsa/dmaap/DMaaPCambriaExceptionMapper.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -35,15 +35,15 @@ import com.att.eelf.configuration.EELFLogger;
 import com.att.eelf.configuration.EELFManager;
 import org.springframework.beans.factory.annotation.Autowired;
 
-import com.att.nsa.cambria.CambriaApiException;
-import com.att.nsa.cambria.exception.DMaaPErrorMessages;
-import com.att.nsa.cambria.exception.DMaaPResponseCode;
-import com.att.nsa.cambria.exception.ErrorResponse;
+import com.att.dmf.mr.CambriaApiException;
+import com.att.dmf.mr.exception.DMaaPErrorMessages;
+import com.att.dmf.mr.exception.DMaaPResponseCode;
+import com.att.dmf.mr.exception.ErrorResponse;
 
 /**
  * Exception Mapper class to handle
  * CambriaApiException 
- * @author author
+ * @author rajashree.khare
  *
  */
 @Provider
diff --git a/src/main/java/com/att/nsa/dmaap/DMaaPWebExceptionMapper.java b/src/main/java/com/att/nsa/dmaap/DMaaPWebExceptionMapper.java
index 47765c3..75c4525 100644
--- a/src/main/java/com/att/nsa/dmaap/DMaaPWebExceptionMapper.java
+++ b/src/main/java/com/att/nsa/dmaap/DMaaPWebExceptionMapper.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -39,14 +39,14 @@ import com.att.eelf.configuration.EELFLogger;
 import com.att.eelf.configuration.EELFManager;
 import org.springframework.beans.factory.annotation.Autowired;
 
-import com.att.nsa.cambria.exception.DMaaPErrorMessages;
-import com.att.nsa.cambria.exception.DMaaPResponseCode;
-import com.att.nsa.cambria.exception.ErrorResponse;
+import com.att.dmf.mr.exception.DMaaPErrorMessages;
+import com.att.dmf.mr.exception.DMaaPResponseCode;
+import com.att.dmf.mr.exception.ErrorResponse;
 
 /**
  * Exception Mapper class to handle
  * Web Exceptions
- * @author author
+ * @author rajashree.khare
  *
  */
 @Provider
@@ -84,7 +84,7 @@ public class DMaaPWebExceptionMapper implements ExceptionMapper<WebApplicationEx
 	 */
 	@Override
 	public Response toResponse(WebApplicationException ex) {
-		
+		//System.out.println("--------------------------------------------------"+ex);
 		LOGGER.info("Reached WebException Mapper");
 		
 		/**
@@ -150,7 +150,7 @@ public class DMaaPWebExceptionMapper implements ExceptionMapper<WebApplicationEx
 		 * Malformed request
 		 */
 		if(ex instanceof BadRequestException)
-		{   ex.printStackTrace();
+		{
 			errRes = new ErrorResponse(HttpStatus.SC_BAD_REQUEST,DMaaPResponseCode.INCORRECT_JSON.
 					getResponseCode(),msgs.getBadRequest());
 			
diff --git a/src/main/java/com/att/nsa/dmaap/HelloWorld.java b/src/main/java/com/att/nsa/dmaap/HelloWorld.java
index 7dc2e0c..a4cccba 100644
--- a/src/main/java/com/att/nsa/dmaap/HelloWorld.java
+++ b/src/main/java/com/att/nsa/dmaap/HelloWorld.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -25,7 +25,7 @@ import org.apache.camel.Exchange;
 
 /**
  * Hello World Sample Camel Service
- * @author author
+ * @author rajashree.khare
  *
  */
 public class HelloWorld {
diff --git a/src/main/java/com/att/nsa/dmaap/JaxrsEchoService.java b/src/main/java/com/att/nsa/dmaap/JaxrsEchoService.java
index 9fcef98..72416da 100644
--- a/src/main/java/com/att/nsa/dmaap/JaxrsEchoService.java
+++ b/src/main/java/com/att/nsa/dmaap/JaxrsEchoService.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -35,7 +35,7 @@ import com.att.nsa.dmaap.filemonitor.ServicePropertiesMap;
 
 /**
  * Example JAX-RS Service
- * @author author
+ * @author rajashree.khare
  *
  */
 @Path("/jaxrs-services")
diff --git a/src/main/java/com/att/nsa/dmaap/JaxrsUserService.java b/src/main/java/com/att/nsa/dmaap/JaxrsUserService.java
index 2724a51..0631a13 100644
--- a/src/main/java/com/att/nsa/dmaap/JaxrsUserService.java
+++ b/src/main/java/com/att/nsa/dmaap/JaxrsUserService.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -30,7 +30,7 @@ import java.util.HashMap;
 
 /**
  * Example JAX-RS User Service
- * @author author
+ * @author rajashree.khare
  *
  */
 @Path("/user")
@@ -39,8 +39,8 @@ public class JaxrsUserService {
 	private static final Map<String,String> userIdToNameMap;
 	static {
 		userIdToNameMap = new HashMap<String,String>();
-		userIdToNameMap.put("user1","User One");
-		userIdToNameMap.put("user2","User Two");
+		userIdToNameMap.put("dw113c","Doug Wait");
+		userIdToNameMap.put("so401q","Stuart O'Day");
 	}
 	
     /**
diff --git a/src/main/java/com/att/nsa/dmaap/filemonitor/ServicePropertiesListener.java b/src/main/java/com/att/nsa/dmaap/filemonitor/ServicePropertiesListener.java
index 8333332..80ff8eb 100644
--- a/src/main/java/com/att/nsa/dmaap/filemonitor/ServicePropertiesListener.java
+++ b/src/main/java/com/att/nsa/dmaap/filemonitor/ServicePropertiesListener.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -26,10 +26,10 @@ import java.io.File;
 //import com.att.ssf.filemonitor.FileChangedListener;
 /**
  * Class ServicePropertiesListener
- * @author author
+ * @author rajashree.khare
  *
  */
-public class ServicePropertiesListener /*implements FileChangedListener*/ {
+public class ServicePropertiesListener/* implements FileChangedListener*/ {
 
 	/**
 	 * Update method
diff --git a/src/main/java/com/att/nsa/dmaap/filemonitor/ServicePropertiesMap.java b/src/main/java/com/att/nsa/dmaap/filemonitor/ServicePropertiesMap.java
index 7f12696..67b9e04 100644
--- a/src/main/java/com/att/nsa/dmaap/filemonitor/ServicePropertiesMap.java
+++ b/src/main/java/com/att/nsa/dmaap/filemonitor/ServicePropertiesMap.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -34,7 +34,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 
 /**
  * ServicePropertiesMap class
- * @author author
+ * @author rajashree.khare
  *
  */
 @SuppressWarnings("squid:S1118") 
@@ -52,27 +52,22 @@ public class ServicePropertiesMap
 	 */
 	public static void refresh(File file) throws Exception
 	{
-		String filePath= null;
 		try
 		{
 			logger.info("Loading properties - " + (file != null?file.getName():""));
 			
 			//Store .json & .properties files into map of maps
-			if (file != null) {
-				filePath = file.getPath();
-			}
+			String filePath = file.getPath();
 			
-			if(filePath != null) {
 			if(filePath.lastIndexOf(".json")>0){
 				
 				ObjectMapper om = new ObjectMapper();
 				TypeReference<HashMap<String, String>> typeRef = 
 						new TypeReference<HashMap<String, String>>() {};
 				HashMap<String, String> propMap = om.readValue(file, typeRef);
-				HashMap<String, String> lcasePropMap = new HashMap<>();
-				for (Map.Entry<String,String> entry : propMap.entrySet())
+				HashMap<String, String> lcasePropMap = new HashMap<String, String>();
+				for (String key : propMap.keySet() )
 				{
-					String key = entry.getKey();
 					String lcaseKey = ifNullThenEmpty(key);
 					lcasePropMap.put(lcaseKey, propMap.get(key));
 				}
@@ -86,11 +81,10 @@ public class ServicePropertiesMap
 				prop.load(fis);
 				
 				@SuppressWarnings("unchecked")
-				HashMap<String, String> propMap = new HashMap<>((Map)prop);
+				HashMap<String, String> propMap = new HashMap<String, String>((Map)prop);
 				
 				mapOfMaps.put(file.getName(), propMap);
 			}
-			}
 
 			logger.info("File - " + file.getName() + " is loaded into the map and the "
 					+ "corresponding system properties have been refreshed");
@@ -117,7 +111,7 @@ public class ServicePropertiesMap
 	 * @param fileName fileName
 	 * @return mapProp
 	 */
-	public static Map<String, String> getProperties(String fileName){
+	public static HashMap<String, String> getProperties(String fileName){
 		return mapOfMaps.get(fileName);
 	}
 	
diff --git a/src/main/java/com/att/nsa/dmaap/filemonitor/ServicePropertyService.java b/src/main/java/com/att/nsa/dmaap/filemonitor/ServicePropertyService.java
index a6a77ba..d573d8b 100644
--- a/src/main/java/com/att/nsa/dmaap/filemonitor/ServicePropertyService.java
+++ b/src/main/java/com/att/nsa/dmaap/filemonitor/ServicePropertyService.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -35,9 +35,12 @@ import com.att.eelf.configuration.EELFLogger;
 import com.att.eelf.configuration.EELFManager;
 
 
+//import com.att.ssf.filemonitor.FileChangedListener;
+//import com.att.ssf.filemonitor.FileMonitor;
+
 /**
  * ServicePropertyService class
- * @author author
+ * @author rajashree.khare
  *
  */
 public class ServicePropertyService {
@@ -65,17 +68,17 @@ public class ServicePropertyService {
 		try {
 			getFileList(FILE_CHANGE_LISTENER_LOC);
 
-//			for (File file : fileList) {
-//					FileChangedListener fileChangedListener = this.fileChangedListener;
-//					Object filePropertiesMap = this.filePropertiesMap;
-//					Method m = filePropertiesMap.getClass().getMethod(
-//							"refresh", File.class);
-//					m.invoke(filePropertiesMap, file);
-//					FileMonitor fm = FileMonitor.getInstance();
-//					fm.addFileChangedListener(file, fileChangedListener,
-//							loadOnStartup);
-//				
-//			}
+			/*for (File file : fileList) {
+					FileChangedListener fileChangedListener = this.fileChangedListener;
+					Object filePropertiesMap = this.filePropertiesMap;
+					Method m = filePropertiesMap.getClass().getMethod(
+							"refresh", File.class);
+					m.invoke(filePropertiesMap, file);
+					FileMonitor fm = FileMonitor.getInstance();
+					fm.addFileChangedListener(file, fileChangedListener,
+							loadOnStartup);
+				
+			}*/
 		} catch (Exception ex) {
 			logger.error("Error creating property map ", ex);
 		}
@@ -87,7 +90,7 @@ public class ServicePropertyService {
 		FileInputStream fis = null;
 
 		if (fileList == null)
-			fileList = new ArrayList<>();
+			fileList = new ArrayList<File>();
 
 		// get all the files that are ".json" or ".properties", from a directory
 		// & it's sub-directories
@@ -107,9 +110,7 @@ public class ServicePropertyService {
 				} catch (Exception ioe) {
 					logger.error("Error reading the file stream ", ioe);
 				} finally {
-					if (fis != null) {
-						fis.close();
-					}
+					fis.close();
 				}
 			} else if (file.isDirectory()) {
 				getFileList(file.getPath());
diff --git a/src/main/java/com/att/nsa/dmaap/mmagent/CreateMirrorMaker.java b/src/main/java/com/att/nsa/dmaap/mmagent/CreateMirrorMaker.java
index 92aca38..4007b17 100644
--- a/src/main/java/com/att/nsa/dmaap/mmagent/CreateMirrorMaker.java
+++ b/src/main/java/com/att/nsa/dmaap/mmagent/CreateMirrorMaker.java
@@ -8,19 +8,29 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
 package com.att.nsa.dmaap.mmagent;
 
+import java.util.Date;
+
+import org.apache.http.HttpStatus;
+import org.json.JSONObject;
+
+import com.att.dmf.mr.CambriaApiException;
+import com.att.dmf.mr.exception.DMaaPResponseCode;
+import com.att.dmf.mr.exception.ErrorResponse;
+import com.att.dmf.mr.utils.Utils;
+
 public class CreateMirrorMaker {
 	String messageID;
 	MirrorMaker createMirrorMaker;
@@ -29,7 +39,7 @@ public class CreateMirrorMaker {
 		return createMirrorMaker;
 	}
 
-	public void setCreateMirrorMaker(MirrorMaker createMirrorMaker) {
+	public void setCreateMirrorMaker(MirrorMaker createMirrorMaker) throws CambriaApiException {
 		this.createMirrorMaker = createMirrorMaker;
 	}
 
@@ -40,4 +50,27 @@ public class CreateMirrorMaker {
 	public void setMessageID(String messageID) {
 		this.messageID = messageID;
 	}
+	public void validateJSON() throws CambriaApiException
+	{
+		ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_BAD_REQUEST,
+				DMaaPResponseCode.INCORRECT_JSON.getResponseCode(), "", null, Utils.getFormattedDate(new Date()), null, null, null, 
+						null,
+				"");
+		if(null==this.createMirrorMaker.getConsumer())
+		{
+			errRes.setErrorMessage("Please provide Consumer host:port details");
+			throw new CambriaApiException(errRes);
+		}
+		if(null==this.createMirrorMaker.getProducer())
+		{
+			errRes.setErrorMessage("Please provide Producer host:port details");
+			throw new CambriaApiException(errRes);
+		}
+		if(this.createMirrorMaker.getNumStreams()<=0)
+		{
+			errRes.setErrorMessage("Please provide numStreams value");
+			throw new CambriaApiException(errRes);
+		}
+		
+	}
 }
diff --git a/src/main/java/com/att/nsa/dmaap/mmagent/MMAgentUtil.java b/src/main/java/com/att/nsa/dmaap/mmagent/MMAgentUtil.java
new file mode 100644
index 0000000..800f82d
--- /dev/null
+++ b/src/main/java/com/att/nsa/dmaap/mmagent/MMAgentUtil.java
@@ -0,0 +1,431 @@
+/**
+ * 
+ */
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ *  ================================================================================
+ *  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.
+ *  ============LICENSE_END=========================================================
+ *  
+ *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ *  
+ *******************************************************************************/
+package com.att.nsa.dmaap.mmagent;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import javax.ws.rs.core.Context;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.http.HttpStatus;
+import org.json.JSONArray;
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.stereotype.Component;
+
+import com.att.ajsc.filemonitor.AJSCPropertiesMap;
+import com.att.dmf.mr.CambriaApiException;
+import com.att.dmf.mr.beans.DMaaPContext;
+import com.att.dmf.mr.constants.CambriaConstants;
+import com.att.dmf.mr.exception.DMaaPErrorMessages;
+import com.att.dmf.mr.exception.DMaaPResponseCode;
+import com.att.dmf.mr.exception.ErrorResponse;
+import com.att.dmf.mr.security.DMaaPAAFAuthenticator;
+import com.att.dmf.mr.security.DMaaPAAFAuthenticatorImpl;
+import com.att.dmf.mr.service.MMService;
+import com.att.dmf.mr.utils.ConfigurationReader;
+import com.att.dmf.mr.utils.DMaaPResponseBuilder;
+import com.att.dmf.mr.utils.Utils;
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+import edu.emory.mathcs.backport.java.util.Arrays;
+
+/**
+ * @author rajashree.khare
+ *Util class for MM Rest Service
+ */
+@Component
+public class MMAgentUtil {/*
+	@Autowired
+	@Qualifier("configurationReader")
+	private ConfigurationReader configReader;
+
+	@Context
+	private HttpServletRequest request;
+
+	@Context
+	private HttpServletResponse response;
+
+	@Autowired
+	private MMService mirrorService;
+
+	private String topic;
+	private int timeout;
+	private String consumergroup;
+	private String consumerid;
+	private static final EELFLogger LOGGER = EELFManager.getInstance().getLogger(MMAgentUtil.class);
+		
+	public JSONObject callPubSub(String randomstr, DMaaPContext ctx, InputStream inStream, String name, boolean listAll) throws Exception {
+		loadProperty();
+		JSONObject jsonObj = new JSONObject();
+		JSONObject finalJsonObj = new JSONObject();
+		JSONArray jsonArray = null;
+		try {
+			String msgFrmSubscribe = mirrorService.subscribe(ctx, topic, consumergroup, consumerid);
+			mirrorService.pushEvents(ctx, topic, inStream, null, null);
+			long startTime = System.currentTimeMillis();
+			
+			while (!isListMirrorMaker(msgFrmSubscribe, randomstr)
+					&& ((System.currentTimeMillis() - startTime) < timeout)) {
+				msgFrmSubscribe = mirrorService.subscribe(ctx, topic, consumergroup, consumerid);
+			
+			}
+			
+
+			if (msgFrmSubscribe != null && msgFrmSubscribe.length() > 0
+					&& isListMirrorMaker(msgFrmSubscribe, randomstr)) {
+				msgFrmSubscribe = removeExtraChar(msgFrmSubscribe);
+				
+				jsonArray = new JSONArray(msgFrmSubscribe);
+				jsonObj = jsonArray.getJSONObject(0);
+				if(jsonObj.has("listMirrorMaker"))
+				{
+					jsonArray = (JSONArray) jsonObj.get("listMirrorMaker");
+					if(true==listAll)
+					{
+						return jsonObj;
+					} 
+					else 
+					{
+						for (int i = 0; i < jsonArray.length(); i++) 
+						{
+							jsonObj = jsonArray.getJSONObject(i);
+								if(null!=name && !name.isEmpty())
+								{
+										if(jsonObj.getString("name").equals(name))
+											{
+												finalJsonObj.put("listMirrorMaker", jsonObj);
+												break;
+											}
+								}
+								else
+								{
+									finalJsonObj.put("listMirrorMaker", jsonObj);
+								}
+						
+						}
+					}
+				}
+				return finalJsonObj;
+
+			} else {
+
+				ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_SERVICE_UNAVAILABLE,
+						DMaaPResponseCode.RESOURCE_NOT_FOUND.getResponseCode(),
+						"listMirrorMaker is not available, please make sure MirrorMakerAgent is running", null,
+						Utils.getFormattedDate(new Date()), topic, null, null, "mirrorMakerAgent", ctx.getRequest().getRemoteHost());
+				LOGGER.info(errRes.toString());
+				throw new CambriaApiException(errRes);
+				
+			}
+			
+		} catch (Exception e) {
+			
+			throw e;
+		}
+	}
+
+	public void sendErrResponse(DMaaPContext ctx, String errMsg) {
+		JSONObject err = new JSONObject();
+		err.append("Error", errMsg);
+
+		try {
+			DMaaPResponseBuilder.respondOk(ctx, err);
+			LOGGER.error(errMsg.toString());
+
+		} catch (JSONException | IOException e) {
+			LOGGER.error(errMsg.toString());
+		}
+	}
+	public boolean isListMirrorMaker(String msg, String messageID) {
+		String topicmsg = msg;
+		topicmsg = removeExtraChar(topicmsg);
+		JSONObject jObj = new JSONObject();
+		JSONArray jArray = null;
+		boolean exist = false;
+
+		if (!StringUtils.isBlank(topicmsg) && topicmsg.length() > 2) {
+			jArray = new JSONArray(topicmsg);
+
+			for (int i = 0; i < jArray.length(); i++) {
+				jObj = jArray.getJSONObject(i);
+				
+				
+				if (jObj.has("messageID") && jObj.get("messageID").equals(messageID) && jObj.has("listMirrorMaker")) {
+					exist = true;
+					break;
+				}
+			}
+		}
+		return exist;
+	}
+
+	public void loadProperty() {
+
+		this.timeout = Integer.parseInt(
+				AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "msgRtr.mirrormaker.timeout").trim());
+		this.topic = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "msgRtr.mirrormaker.topic").trim();
+		this.consumergroup = AJSCPropertiesMap
+				.getProperty(CambriaConstants.msgRtr_prop, "msgRtr.mirrormaker.consumergroup").trim();
+		this.consumerid = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "msgRtr.mirrormaker.consumerid")
+				.trim();
+		
+
+	}
+
+	public String removeExtraChar(String message) {
+		String str = message;
+		str = checkJsonFormate(str);
+
+		if (str != null && str.length() > 0) {
+			str = str.replace("\\", "");
+			str = str.replace("\"{", "{");
+			str = str.replace("}\"", "}");
+		}
+		return str;
+	}
+
+	public String getRandomNum() {
+		long random = Math.round(Math.random() * 89999) + 10000;
+		String strLong = Long.toString(random);
+		return strLong;
+	}
+
+	public boolean isAlphaNumeric(String name) {
+		String pattern = "^[a-zA-Z0-9]*$";
+		if (name.matches(pattern)) {
+			return true;
+		}
+		return false;
+	}
+
+	// This method validate IPv4
+	public boolean validateIPPort(String ipPort) {
+		String pattern = "^([01]?\\d\\d?|2[0-4]\\d|25[0-5])\\.([01]?\\d\\d?|2[0-4]\\d|25[0-5])\\."
+				+ "([01]?\\d\\d?|2[0-4]\\d|25[0-5])\\.([01]?\\d\\d?|2[0-4]\\d|25[0-5]):"
+				+ "([1-9][0-9]{0,3}|[1-5][0-9]{4}|6[0-4][0-9]{3}|65[0-4][0-9]{2}|655[0-2][0-9]|6553[0-5])$";
+		if (ipPort.matches(pattern)) {
+			return true;
+		}
+		return false;
+	}
+
+	public String checkJsonFormate(String jsonStr) {
+
+		String json = jsonStr;
+		if (jsonStr != null && jsonStr.length() > 0 && jsonStr.startsWith("[") && !jsonStr.endsWith("]")) {
+			json = json + "]";
+		}
+		return json;
+	}
+
+	public boolean checkMirrorMakerPermission(DMaaPContext ctx, String permission) {
+
+		boolean hasPermission = false;
+
+		DMaaPAAFAuthenticator aaf = new DMaaPAAFAuthenticatorImpl();
+
+		if (aaf.aafAuthentication(ctx.getRequest(), permission)) {
+			hasPermission = true;
+		}
+		return hasPermission;
+	}
+	
+	public String getNamespace(String topic) {
+		return topic.substring(0, topic.lastIndexOf("."));
+	}
+
+	public String removeTopic(String whitelist, String topicToRemove) {
+		List<String> topicList = new ArrayList<String>();
+		List<String> newTopicList = new ArrayList<String>();
+
+		if (whitelist.contains(",")) {
+			topicList = Arrays.asList(whitelist.split(","));
+
+		}
+
+		if (topicList.contains(topicToRemove)) {
+			for (String topic : topicList) {
+				if (!topic.equals(topicToRemove)) {
+					newTopicList.add(topic);
+				}
+			}
+		}
+
+		String newWhitelist = StringUtils.join(newTopicList, ",");
+
+		return newWhitelist;
+	}
+
+	public void callPubSubForWhitelist(String randomStr, DMaaPContext ctx, InputStream inStream, JSONObject jsonOb) {
+		
+		loadProperty();
+		try {
+			String namespace = jsonOb.getString("namespace");
+			String mmName = jsonOb.getString("name");
+			
+			String msgFrmSubscribe = mirrorService.subscribe(ctx, topic, consumergroup, consumerid);
+			mirrorService.pushEvents(ctx, topic, inStream, null, null);
+			long startTime = System.currentTimeMillis();
+
+			while (!isListMirrorMaker(msgFrmSubscribe, randomStr)
+					&& (System.currentTimeMillis() - startTime) < timeout) {
+				msgFrmSubscribe = mirrorService.subscribe(ctx, topic, consumergroup, consumerid);
+			}
+			
+			JSONObject jsonObj = new JSONObject();
+			JSONArray jsonArray = null;
+			JSONArray jsonArrayNamespace = null;
+
+			if (msgFrmSubscribe != null && msgFrmSubscribe.length() > 0
+					&& isListMirrorMaker(msgFrmSubscribe, randomStr)) {
+				msgFrmSubscribe = removeExtraChar(msgFrmSubscribe);
+				jsonArray = new JSONArray(msgFrmSubscribe);
+
+				for (int i = 0; i < jsonArray.length(); i++) {
+					jsonObj = jsonArray.getJSONObject(i);
+					
+
+					if (jsonObj.has("messageID") && jsonObj.get("messageID").equals(randomStr) && jsonObj.has("listMirrorMaker")) {
+						jsonArrayNamespace = jsonObj.getJSONArray("listMirrorMaker");
+					}
+				}
+				JSONObject finalJasonObj = new JSONObject();
+				JSONArray finalJsonArray = new JSONArray();
+
+				for (int i = 0; i < jsonArrayNamespace.length(); i++) {
+
+					JSONObject mmObj = new JSONObject();
+					mmObj = jsonArrayNamespace.getJSONObject(i);
+					if(mmObj.has("name")&& mmName.equals(mmObj.getString("name")))
+					{
+										
+						finalJsonArray.put(mmObj);
+					}
+					
+				}
+				finalJasonObj.put("listMirrorMaker", finalJsonArray);
+
+				DMaaPResponseBuilder.respondOk(ctx, finalJasonObj);
+
+			} else {
+
+				ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_SERVICE_UNAVAILABLE,
+						DMaaPResponseCode.RESOURCE_NOT_FOUND.getResponseCode(),
+						"listMirrorMaker is not available, please make sure MirrorMakerAgent is running", null,
+						Utils.getFormattedDate(new Date()), topic, null, null, "mirrorMakerAgent", ctx.getRequest().getRemoteHost());
+				LOGGER.info(errRes.toString());
+				throw new CambriaApiException(errRes);
+			}
+
+		} catch (Exception e) {
+			e.printStackTrace();
+		}
+	}
+
+	public String getWhitelistByNamespace(String originalWhitelist, String namespace) {
+
+		String whitelist = null;
+		List<String> resultList = new ArrayList<String>();
+		List<String> whitelistList = new ArrayList<String>();
+		whitelistList = Arrays.asList(originalWhitelist.split(","));
+
+		for (String topic : whitelistList) {
+			if (StringUtils.isNotBlank(originalWhitelist) && getNamespace(topic).equals(namespace)) {
+				resultList.add(topic);
+			}
+		}
+		if (resultList.size() > 0) {
+			whitelist = StringUtils.join(resultList, ",");
+		}
+
+		return whitelist;
+	}
+	
+	public JSONArray getListMirrorMaker(String msgFrmSubscribe, String randomStr) {
+		JSONObject jsonObj = new JSONObject();
+		JSONArray jsonArray = new JSONArray();
+		JSONArray listMirrorMaker = new JSONArray();
+		
+		msgFrmSubscribe = removeExtraChar(msgFrmSubscribe);
+		jsonArray = new JSONArray(msgFrmSubscribe);
+		jsonObj = jsonArray.getJSONObject(0);
+		
+		for (int i = 0; i < jsonArray.length(); i++) {
+			jsonObj = jsonArray.getJSONObject(i);
+			
+			if (jsonObj.has("messageID") && jsonObj.get("messageID").equals(randomStr) && jsonObj.has("listMirrorMaker")) {
+				listMirrorMaker = jsonObj.getJSONArray("listMirrorMaker");
+				break;
+			}
+		}
+		return listMirrorMaker;		
+	}
+	
+	public  JSONObject validateMMExists(DMaaPContext ctx,String name) throws Exception
+	{
+		// Create a listAllMirrorMaker Json object
+		JSONObject listAll = new JSONObject();
+		try {
+			listAll.put("listAllMirrorMaker", new JSONObject());
+
+		} catch (JSONException e) {
+
+			e.printStackTrace();
+		}
+
+		// set a random number as messageID
+		String randomStr = getRandomNum();
+		listAll.put("messageID", randomStr);
+		InputStream inStream = null;
+
+		// convert listAll Json object to InputStream object
+		try {
+			inStream = IOUtils.toInputStream(listAll.toString(), "UTF-8");
+
+		} catch (IOException ioe) {
+			ioe.printStackTrace();
+		}
+		JSONObject listMirrorMaker =new JSONObject();
+		listMirrorMaker = callPubSub(randomStr, ctx, inStream, name, false);
+		if (null!=listMirrorMaker && listMirrorMaker.length()>0){
+			listMirrorMaker.put("exists", true);
+			return listMirrorMaker;
+			
+		}
+		listMirrorMaker.put("exists", false);
+		return listMirrorMaker;
+	
+	}
+*/}
diff --git a/src/main/java/com/att/nsa/dmaap/mmagent/MirrorMaker.java b/src/main/java/com/att/nsa/dmaap/mmagent/MirrorMaker.java
index f9e6d89..cb19a00 100644
--- a/src/main/java/com/att/nsa/dmaap/mmagent/MirrorMaker.java
+++ b/src/main/java/com/att/nsa/dmaap/mmagent/MirrorMaker.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -27,6 +27,8 @@ public class MirrorMaker {
 	public String producer;
 	public String whitelist;
 	public String status;
+	public int numStreams =1;
+	public boolean enablelogCheck = false;
 
 	public String getStatus() {
 		return status;
@@ -67,4 +69,20 @@ public class MirrorMaker {
 	public void setWhitelist(String whitelist) {
 		this.whitelist = whitelist;
 	}
+
+	public int getNumStreams() {
+		return numStreams;
+	}
+
+	public void setNumStreams(int numStreams) {
+		this.numStreams = numStreams;
+	}
+
+	public boolean isEnablelogCheck() {
+		return enablelogCheck;
+	}
+
+	public void setEnablelogCheck(boolean enablelogCheck) {
+		this.enablelogCheck = enablelogCheck;
+	}
 }
\ No newline at end of file
diff --git a/src/main/java/com/att/nsa/dmaap/mmagent/UpdateMirrorMaker.java b/src/main/java/com/att/nsa/dmaap/mmagent/UpdateMirrorMaker.java
index 4d291f3..4a0a4b6 100644
--- a/src/main/java/com/att/nsa/dmaap/mmagent/UpdateMirrorMaker.java
+++ b/src/main/java/com/att/nsa/dmaap/mmagent/UpdateMirrorMaker.java
@@ -8,19 +8,29 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
 package com.att.nsa.dmaap.mmagent;
 
+import java.util.Date;
+
+import org.apache.http.HttpStatus;
+import org.json.JSONObject;
+
+import com.att.dmf.mr.CambriaApiException;
+import com.att.dmf.mr.exception.DMaaPResponseCode;
+import com.att.dmf.mr.exception.ErrorResponse;
+import com.att.dmf.mr.utils.Utils;
+
 public class UpdateMirrorMaker {
 	String messageID;
 	MirrorMaker updateMirrorMaker;
@@ -40,4 +50,34 @@ public class UpdateMirrorMaker {
 	public void setMessageID(String messageID) {
 		this.messageID = messageID;
 	}
+	public void validateJSON(JSONObject jsonObj) throws CambriaApiException
+	{
+		ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_BAD_REQUEST,
+				DMaaPResponseCode.INCORRECT_JSON.getResponseCode(), "", null, Utils.getFormattedDate(new Date()), null, null, null, 
+						null,
+				"");
+
+	
+		if(jsonObj.has("consumer")&& null==this.updateMirrorMaker.getConsumer())
+		{
+			errRes.setErrorMessage("Please provide Consumer host:port details");
+			throw new CambriaApiException(errRes);
+		}
+		if(jsonObj.has("producer")&& null==this.updateMirrorMaker.getProducer())
+		{
+			errRes.setErrorMessage("Please provide Producer host:port details");
+			throw new CambriaApiException(errRes);
+		}
+		if(jsonObj.has("numStreams")&& this.updateMirrorMaker.getNumStreams()<=0)
+		{
+			errRes.setErrorMessage("Please provide numStreams value");
+			throw new CambriaApiException(errRes);
+		}
+		if(jsonObj.has("whitelist"))
+		{
+			errRes.setErrorMessage("Please use Create Whitelist API to add whitelist topics");
+			throw new CambriaApiException(errRes);
+		}
+	
+	}
 }
diff --git a/src/main/java/com/att/nsa/dmaap/mmagent/UpdateWhiteList.java b/src/main/java/com/att/nsa/dmaap/mmagent/UpdateWhiteList.java
index 616dc85..a1064a4 100644
--- a/src/main/java/com/att/nsa/dmaap/mmagent/UpdateWhiteList.java
+++ b/src/main/java/com/att/nsa/dmaap/mmagent/UpdateWhiteList.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
diff --git a/src/main/java/com/att/nsa/dmaap/service/AdminRestService.java b/src/main/java/com/att/nsa/dmaap/service/AdminRestService.java
index 2ab574d..49e752a 100644
--- a/src/main/java/com/att/nsa/dmaap/service/AdminRestService.java
+++ b/src/main/java/com/att/nsa/dmaap/service/AdminRestService.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -44,19 +44,19 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Component;
 
-import com.att.nsa.cambria.CambriaApiException;
-import com.att.nsa.cambria.beans.DMaaPContext;
-import com.att.nsa.cambria.exception.DMaaPResponseCode;
-import com.att.nsa.cambria.exception.ErrorResponse;
-import com.att.nsa.cambria.service.AdminService;
-import com.att.nsa.cambria.utils.ConfigurationReader;
+import com.att.dmf.mr.CambriaApiException;
+import com.att.dmf.mr.beans.DMaaPContext;
+import com.att.dmf.mr.exception.DMaaPResponseCode;
+import com.att.dmf.mr.exception.ErrorResponse;
+import com.att.dmf.mr.service.AdminService;
+import com.att.dmf.mr.utils.ConfigurationReader;
 import com.att.nsa.configs.ConfigDbException;
 import com.att.nsa.security.ReadWriteSecuredResource.AccessDeniedException;
 
 /**
  * Rest Service class
  * for Admin Services
- * @author author
+ * @author Ramkumar
  *
  */
 @Component
@@ -91,8 +91,6 @@ public class AdminRestService {
 	 */
 	@Autowired
 	private AdminService adminService;
-	
-	private DMaaPContext dmaaPContext;
 
 	/**
 	 * Fetches a list of all the registered consumers along with their created
@@ -172,7 +170,7 @@ public class AdminRestService {
 	public void getBlacklist() throws CambriaApiException {
 		LOGGER.info("Fetching list of blacklist ips.");
 		try {
-			Enumeration headerNames = ServiceUtil.getDMaaPContext(configReader, request, response).getRequest().getHeaderNames();
+			Enumeration headerNames =ServiceUtil.getDMaaPContext(configReader, request, response).getRequest().getHeaderNames();
 			while (headerNames.hasMoreElements()) {
 				String key = (String) headerNames.nextElement();
 				String value = request.getHeader(key);
@@ -280,6 +278,5 @@ public class AdminRestService {
 		}
 	}
 
-	
 
 }
diff --git a/src/main/java/com/att/nsa/dmaap/service/ApiKeysRestService.java b/src/main/java/com/att/nsa/dmaap/service/ApiKeysRestService.java
index a76a04c..2dea889 100644
--- a/src/main/java/com/att/nsa/dmaap/service/ApiKeysRestService.java
+++ b/src/main/java/com/att/nsa/dmaap/service/ApiKeysRestService.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -43,13 +43,13 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Component;
 
-import com.att.nsa.cambria.CambriaApiException;
-import com.att.nsa.cambria.beans.ApiKeyBean;
-import com.att.nsa.cambria.beans.DMaaPContext;
-import com.att.nsa.cambria.exception.DMaaPResponseCode;
-import com.att.nsa.cambria.exception.ErrorResponse;
-import com.att.nsa.cambria.service.ApiKeysService;
-import com.att.nsa.cambria.utils.ConfigurationReader;
+import com.att.dmf.mr.CambriaApiException;
+import com.att.dmf.mr.beans.ApiKeyBean;
+import com.att.dmf.mr.beans.DMaaPContext;
+import com.att.dmf.mr.exception.DMaaPResponseCode;
+import com.att.dmf.mr.exception.ErrorResponse;
+import com.att.dmf.mr.service.ApiKeysService;
+import com.att.dmf.mr.utils.ConfigurationReader;
 import com.att.nsa.configs.ConfigDbException;
 import com.att.nsa.security.db.NsaApiDb.KeyExistsException;
 import com.att.nsa.security.ReadWriteSecuredResource.AccessDeniedException;
@@ -58,7 +58,7 @@ import com.att.nsa.security.ReadWriteSecuredResource.AccessDeniedException;
  * This class is a CXF REST service 
  * which acts as gateway for Cambria Api
  * Keys.
- * @author author
+ * @author rajashree.khare
  *
  */
 @Component
@@ -107,7 +107,7 @@ public class ApiKeysRestService {
 		log.info("Inside ApiKeysRestService.getAllApiKeys");
 
 		try {
-			apiKeyService.getAllApiKeys(ServiceUtil.getDMaaPContext(configReader, request, response));
+			apiKeyService.getAllApiKeys(getDmaapContext());
 			log.info("Fetching all API keys is Successful");
 		} catch (ConfigDbException | IOException e) {
 			log.error("Error while retrieving API keys: " + e);
@@ -135,7 +135,7 @@ public class ApiKeysRestService {
 		log.info("Fetching details of api key: " + apiKeyName);
 
 		try {
-			apiKeyService.getApiKey(ServiceUtil.getDMaaPContext(configReader, request, response), apiKeyName);
+			apiKeyService.getApiKey(getDmaapContext(), apiKeyName);
 			log.info("Fetching specific API key is Successful");
 		} catch (ConfigDbException | IOException e) {
 			log.error("Error while retrieving API key details: " + e);
@@ -160,11 +160,11 @@ public class ApiKeysRestService {
 	@POST
 	@Path("/create")
 	@Consumes(MediaType.APPLICATION_JSON)
-	public void createApiKey(ApiKeyBean nsaApiKey) throws CambriaApiException {
+	public void createApiKey(ApiKeyBean nsaApiKey) throws CambriaApiException, JSONException {
 		log.info("Creating Api Key.");
 
 		try {
-			apiKeyService.createApiKey(ServiceUtil.getDMaaPContext(configReader, request, response), nsaApiKey);
+			apiKeyService.createApiKey(getDmaapContext(), nsaApiKey);
 			log.info("Creating API key is Successful");
 		} catch (KeyExistsException | ConfigDbException | IOException e) {
 			log.error("Error while Creating API key : " + e.getMessage(), e);
@@ -192,13 +192,13 @@ public class ApiKeysRestService {
 	@PUT
 	@Path("/{apiKey}")
 	public void updateApiKey(@PathParam("apiKey") String apiKeyName,
-			ApiKeyBean nsaApiKey) throws CambriaApiException  {
+			ApiKeyBean nsaApiKey) throws CambriaApiException, JSONException {
 		log.info("Updating Api Key.");
 
 		try {
 			
 			apiKeyService
-					.updateApiKey(ServiceUtil.getDMaaPContext(configReader, request, response), apiKeyName, nsaApiKey);
+					.updateApiKey(getDmaapContext(), apiKeyName, nsaApiKey);
 			log.error("API key updated sucessfully");
 		} catch (ConfigDbException | IOException | AccessDeniedException e) {
 			log.error("Error while Updating API key : " + apiKeyName, e);
@@ -225,7 +225,7 @@ public class ApiKeysRestService {
 	public void deleteApiKey(@PathParam("apiKey") String apiKeyName) throws CambriaApiException {
 		log.info("Deleting Api Key: " + apiKeyName);
 		try {
-			apiKeyService.deleteApiKey(ServiceUtil.getDMaaPContext(configReader, request, response), apiKeyName);
+			apiKeyService.deleteApiKey(getDmaapContext(), apiKeyName);
 			log.info("Api Key deleted successfully: " + apiKeyName);
 		} catch (ConfigDbException | IOException | AccessDeniedException e) {
 			log.error("Error while deleting API key : " + apiKeyName, e);
@@ -239,5 +239,16 @@ public class ApiKeysRestService {
 		}
 	}
 
+	/**
+	 * Create a dmaap context
+	 * @return DMaaPContext
+	 */
+	private DMaaPContext getDmaapContext() {
+		DMaaPContext dmaapContext = new DMaaPContext();
+		dmaapContext.setConfigReader(configReader);
+		dmaapContext.setRequest(request);
+		dmaapContext.setResponse(response);
+		return dmaapContext;
+	}
 
 }
\ No newline at end of file
diff --git a/src/main/java/com/att/nsa/dmaap/service/EventsRestService.java b/src/main/java/com/att/nsa/dmaap/service/EventsRestService.java
index 6fbfd01..40468a3 100644
--- a/src/main/java/com/att/nsa/dmaap/service/EventsRestService.java
+++ b/src/main/java/com/att/nsa/dmaap/service/EventsRestService.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -42,25 +42,24 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Component;
 
-import com.att.nsa.cambria.CambriaApiException;
-import com.att.nsa.cambria.backends.ConsumerFactory.UnavailableException;
-import com.att.nsa.cambria.beans.DMaaPContext;
-import com.att.nsa.cambria.exception.DMaaPErrorMessages;
-import com.att.nsa.cambria.exception.DMaaPResponseCode;
-import com.att.nsa.cambria.exception.ErrorResponse;
-import com.att.nsa.cambria.metabroker.Broker.TopicExistsException;
-import com.att.nsa.cambria.service.EventsService;
-import com.att.nsa.cambria.utils.ConfigurationReader;
-import com.att.nsa.cambria.utils.Utils;
+import com.att.dmf.mr.CambriaApiException;
+import com.att.dmf.mr.backends.ConsumerFactory.UnavailableException;
+import com.att.dmf.mr.beans.DMaaPContext;
+import com.att.dmf.mr.exception.DMaaPErrorMessages;
+import com.att.dmf.mr.exception.DMaaPResponseCode;
+import com.att.dmf.mr.exception.ErrorResponse;
+import com.att.dmf.mr.metabroker.Broker.TopicExistsException;
+import com.att.dmf.mr.service.EventsService;
+import com.att.dmf.mr.utils.ConfigurationReader;
+import com.att.dmf.mr.utils.Utils;
 import com.att.nsa.configs.ConfigDbException;
 import com.att.nsa.drumlin.till.nv.rrNvReadable.missingReqdSetting;
 import com.att.nsa.security.ReadWriteSecuredResource.AccessDeniedException;
-import com.att.nsa.cambria.exception.DMaaPAccessDeniedException;
-
+import com.att.dmf.mr.exception.DMaaPAccessDeniedException;
 /**
- * This class is a CXF REST service which acts as gateway for MR Event Service.
- * 
- * @author author
+ * This class is a CXF REST service which acts 
+ * as gateway for MR Event Service.
+ * @author rajashree.khare
  *
  */
 @Component
@@ -70,8 +69,7 @@ public class EventsRestService {
 	/**
 	 * Logger obj
 	 */
-	// private Logger log =
-	// Logger.getLogger(EventsRestService.class.toString());
+	//private Logger log = Logger.getLogger(EventsRestService.class.toString());
 	private static final EELFLogger log = EELFManager.getInstance().getLogger(EventsRestService.class);
 	/**
 	 * HttpServletRequest obj
@@ -85,6 +83,7 @@ public class EventsRestService {
 	@Context
 	private HttpServletResponse response;
 
+
 	/**
 	 * Config Reader
 	 */
@@ -98,8 +97,6 @@ public class EventsRestService {
 	@Autowired
 	private DMaaPErrorMessages errorMessages;
 
-	private DMaaPContext dmaapContext = new DMaaPContext();
-
 	/**
 	 * This method is used to consume messages.Taking three parameter
 	 * topic,consumerGroup and consumerId .Consumer decide to which topic they
@@ -121,49 +118,120 @@ public class EventsRestService {
 	 */
 	@GET
 	@Path("/{topic}/{consumergroup}/{consumerid}")
-	public void getEvents(@PathParam("topic") String topic, @PathParam("consumergroup") String consumergroup,
+	public void getEvents(@PathParam("topic") String topic, @PathParam("consumergroup") 
+	String consumergroup,
 			@PathParam("consumerid") String consumerid) throws CambriaApiException {
 		// log.info("Consuming message from topic " + topic );
-		dmaapContext = getDmaapContext();
-		dmaapContext.setConsumerRequestTime(Utils.getFormattedDate(new Date()));
+		DMaaPContext dMaaPContext = getDmaapContext();
+		dMaaPContext.setConsumerRequestTime(Utils.getFormattedDate(new Date()));
 
 		try {
 
-			eventsService.getEvents(dmaapContext, topic, consumergroup, consumerid);
-		} catch (TopicExistsException e) {
+			eventsService.getEvents(dMaaPContext, topic, consumergroup, consumerid);
+		} 
+		catch (TopicExistsException  e) {
 			log.error("Error while reading data from topic [" + topic + "].", e);
 
 			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_CONFLICT,
-					DMaaPResponseCode.CONSUME_MSG_ERROR.getResponseCode(),
-					errorMessages.getConsumeMsgError() + e.getMessage(), null, Utils.getFormattedDate(new Date()),
-					topic, null, null, consumerid, request.getRemoteHost());
+					DMaaPResponseCode.CONSUME_MSG_ERROR.getResponseCode(), errorMessages.getConsumeMsgError()
+							+ e.getMessage(), null, Utils.getFormattedDate(new Date()), topic, null, null, 
+							consumerid,
+					request.getRemoteHost());
 			log.info(errRes.toString());
 			throw new CambriaApiException(errRes);
 
-		} catch (DMaaPAccessDeniedException | AccessDeniedException e) {
+		}
+		catch (DMaaPAccessDeniedException | AccessDeniedException  e) {
 			log.error("Error while reading data from topic [" + topic + "].", e);
 
 			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_UNAUTHORIZED,
-					DMaaPResponseCode.CONSUME_MSG_ERROR.getResponseCode(),
-					errorMessages.getConsumeMsgError() + e.getMessage(), null, Utils.getFormattedDate(new Date()),
-					topic, null, null, consumerid, request.getRemoteHost());
+					DMaaPResponseCode.CONSUME_MSG_ERROR.getResponseCode(), errorMessages.getConsumeMsgError()
+							+ e.getMessage(), null, Utils.getFormattedDate(new Date()), topic, null, null, 
+							consumerid,
+					request.getRemoteHost());
 			log.info(errRes.toString());
 			throw new CambriaApiException(errRes);
 
 		}
-
+		
 		catch (ConfigDbException | UnavailableException | IOException e) {
 			log.error("Error while reading data from topic [" + topic + "].", e);
+		
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND,
+					DMaaPResponseCode.CONSUME_MSG_ERROR.getResponseCode(), errorMessages.getConsumeMsgError()
+							+ e.getMessage(), null, Utils.getFormattedDate(new Date()), topic, null, null, 
+							consumerid,
+					request.getRemoteHost());
+			log.info(errRes.toString());
+			throw new CambriaApiException(errRes);
+
+		}
+	}
+	
+	
+	/**
+	 * This method is used to throw an exception back to the client app if CG/CID is not passed
+	 *  while consuming messages
+	 */
+	@GET
+	@Path("/{topic}")
+	public void getEventsToException(@PathParam("topic") String topic) throws CambriaApiException {
+		// log.info("Consuming message from topic " + topic );
+		DMaaPContext dMaaPContext = getDmaapContext();
+		dMaaPContext.setConsumerRequestTime(Utils.getFormattedDate(new Date()));
+
+		try {
+
+			throw new TopicExistsException("Incorrect URL");
+		} 
+		catch (TopicExistsException  e) {
+			log.error("Error while reading data from topic [" + topic + "].", e);
+
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND,
+					DMaaPResponseCode.CONSUME_MSG_ERROR.getResponseCode(), "Incorrect url - Expects consumer Group and ID in " + request.getRequestURI() + " from "+request.getRemoteHost()
+					);
+			log.info(errRes.toString());
+			throw new CambriaApiException(errRes);
+
+		}
+		
+	}
+	
+	/**
+	 * This method is used to throw an exception back to the client app if CG/CID is not passed
+	 *  while consuming messages
+	 */
+	@GET
+	@Path("/{topic}/{consumergroup}")
+	public void getEventsToException(@PathParam("topic") String topic, @PathParam("consumergroup") 
+	String consumergroup
+			) throws CambriaApiException {
+		// log.info("Consuming message from topic " + topic );
+		DMaaPContext dMaaPContext = getDmaapContext();
+		dMaaPContext.setConsumerRequestTime(Utils.getFormattedDate(new Date()));
+
+		try {
+
+			throw new TopicExistsException("Incorrect URL");
+		} 
+		catch (TopicExistsException  e) {
+			log.error("Error while reading data from topic [" + topic + "].", e);
 
 			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND,
-					DMaaPResponseCode.CONSUME_MSG_ERROR.getResponseCode(),
-					errorMessages.getConsumeMsgError() + e.getMessage(), null, Utils.getFormattedDate(new Date()),
-					topic, null, null, consumerid, request.getRemoteHost());
+					DMaaPResponseCode.CONSUME_MSG_ERROR.getResponseCode(), "Incorrect url - Expects consumer ID in " + request.getRequestURI() + " from "+request.getRemoteHost()
+					);
 			log.info(errRes.toString());
 			throw new CambriaApiException(errRes);
 
 		}
+		
 	}
+	
+	
+	
+	
+	
+	
 
 	/**
 	 * This method is used to publish messages.Taking two parameter topic and
@@ -189,33 +257,36 @@ public class EventsRestService {
 
 		try {
 			eventsService.pushEvents(getDmaapContext(), topic, msg, partitionKey, null);
-		} catch (TopicExistsException e) {
+		} 
+		catch ( TopicExistsException  e) {
 			log.error("Error while publishing to topic [" + topic + "].", e);
 
 			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_CONFLICT,
-					DMaaPResponseCode.PUBLISH_MSG_ERROR.getResponseCode(),
-					errorMessages.getPublishMsgError() + e.getMessage(), null, Utils.getFormattedDate(new Date()),
-					topic, Utils.getUserApiKey(request), request.getRemoteHost(), null, null);
+					DMaaPResponseCode.PUBLISH_MSG_ERROR.getResponseCode(), errorMessages.getPublishMsgError()
+							+ e.getMessage(), null, Utils.getFormattedDate(new Date()), topic,
+					Utils.getUserApiKey(request), request.getRemoteHost(), null, null);
 			log.info(errRes.toString());
 			throw new CambriaApiException(errRes);
-		} catch (DMaaPAccessDeniedException | AccessDeniedException e) {
+		}
+		catch ( DMaaPAccessDeniedException | AccessDeniedException  e) {
 			log.error("Error while publishing to topic [" + topic + "].", e);
 
 			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_UNAUTHORIZED,
-					DMaaPResponseCode.PUBLISH_MSG_ERROR.getResponseCode(),
-					errorMessages.getPublishMsgError() + e.getMessage(), null, Utils.getFormattedDate(new Date()),
-					topic, Utils.getUserApiKey(request), request.getRemoteHost(), null, null);
+					DMaaPResponseCode.PUBLISH_MSG_ERROR.getResponseCode(), errorMessages.getPublishMsgError()
+							+ e.getMessage(), null, Utils.getFormattedDate(new Date()), topic,
+					Utils.getUserApiKey(request), request.getRemoteHost(), null, null);
 			log.info(errRes.toString());
 			throw new CambriaApiException(errRes);
 		}
-
-		catch (ConfigDbException | IOException | missingReqdSetting e) {
+		
+		
+		catch (ConfigDbException |   IOException | missingReqdSetting e) {
 			log.error("Error while publishing to topic [" + topic + "].", e);
 
 			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND,
-					DMaaPResponseCode.PUBLISH_MSG_ERROR.getResponseCode(),
-					errorMessages.getPublishMsgError() + e.getMessage(), null, Utils.getFormattedDate(new Date()),
-					topic, Utils.getUserApiKey(request), request.getRemoteHost(), null, null);
+					DMaaPResponseCode.PUBLISH_MSG_ERROR.getResponseCode(), errorMessages.getPublishMsgError()
+							+ e.getMessage(), null, Utils.getFormattedDate(new Date()), topic,
+					Utils.getUserApiKey(request), request.getRemoteHost(), null, null);
 			log.info(errRes.toString());
 			throw new CambriaApiException(errRes);
 		}
@@ -244,37 +315,40 @@ public class EventsRestService {
 		// );
 
 		try {
-			eventsService.pushEvents(getDmaapContext(), topic, request.getInputStream(), partitionKey,
+			eventsService.pushEvents(getDmaapContext(), topic, request.getInputStream(), 
+					partitionKey,
 					Utils.getFormattedDate(new Date()));
-		}
-
-		catch (TopicExistsException e) {
+		} 
+		
+		catch ( TopicExistsException  e) {
 			log.error("Error while publishing to topic [" + topic + "].", e);
 
 			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_CONFLICT,
-					DMaaPResponseCode.PUBLISH_MSG_ERROR.getResponseCode(),
-					errorMessages.getPublishMsgError() + e.getMessage(), null, Utils.getFormattedDate(new Date()),
-					topic, Utils.getUserApiKey(request), request.getRemoteHost(), null, null);
+					DMaaPResponseCode.PUBLISH_MSG_ERROR.getResponseCode(), errorMessages.getPublishMsgError()
+							+ e.getMessage(), null, Utils.getFormattedDate(new Date()), topic,
+					Utils.getUserApiKey(request), request.getRemoteHost(), null, null);
 			log.info(errRes.toString());
 			throw new CambriaApiException(errRes);
-		} catch (DMaaPAccessDeniedException | AccessDeniedException e) {
+		}
+		catch ( DMaaPAccessDeniedException| AccessDeniedException  e) {
 			log.error("Error while publishing to topic [" + topic + "].", e);
 
 			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_UNAUTHORIZED,
-					DMaaPResponseCode.PUBLISH_MSG_ERROR.getResponseCode(),
-					errorMessages.getPublishMsgError() + e.getMessage(), null, Utils.getFormattedDate(new Date()),
-					topic, Utils.getUserApiKey(request), request.getRemoteHost(), null, null);
+					DMaaPResponseCode.PUBLISH_MSG_ERROR.getResponseCode(), errorMessages.getPublishMsgError()
+							+ e.getMessage(), null, Utils.getFormattedDate(new Date()), topic,
+					Utils.getUserApiKey(request), request.getRemoteHost(), null, null);
 			log.info(errRes.toString());
 			throw new CambriaApiException(errRes);
 		}
-
-		catch (ConfigDbException | IOException | missingReqdSetting e) {
+		
+		catch (ConfigDbException  | IOException | missingReqdSetting  e) {
 			log.error("Error while publishing to topic : " + topic, e);
 
 			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND,
-					DMaaPResponseCode.PUBLISH_MSG_ERROR.getResponseCode(),
-					"Transaction-" + errorMessages.getPublishMsgError() + e.getMessage(), null,
-					Utils.getFormattedDate(new Date()), topic, Utils.getUserApiKey(request), request.getRemoteHost(),
+					DMaaPResponseCode.PUBLISH_MSG_ERROR.getResponseCode(), "Transaction-"
+							+ errorMessages.getPublishMsgError() + e.getMessage(), null,
+					Utils.getFormattedDate(new Date()), topic, Utils.getUserApiKey(request), 
+					request.getRemoteHost(),
 					null, null);
 			log.info(errRes.toString());
 			throw new CambriaApiException(errRes);
@@ -293,6 +367,7 @@ public class EventsRestService {
 	 */
 	private DMaaPContext getDmaapContext() {
 
+		DMaaPContext dmaapContext = new DMaaPContext();
 		dmaapContext.setRequest(request);
 		dmaapContext.setResponse(response);
 		dmaapContext.setConfigReader(configReader);
diff --git a/src/main/java/com/att/nsa/dmaap/service/MMRestService.java b/src/main/java/com/att/nsa/dmaap/service/MMRestService.java
index e816da5..0415c57 100644
--- a/src/main/java/com/att/nsa/dmaap/service/MMRestService.java
+++ b/src/main/java/com/att/nsa/dmaap/service/MMRestService.java
@@ -42,9 +42,9 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Component;
 
-import com.att.nsa.cambria.utils.ConfigurationReader;
-import com.att.nsa.cambria.utils.DMaaPResponseBuilder;
-import com.att.nsa.cambria.utils.Utils;
+import com.att.dmf.mr.utils.ConfigurationReader;
+import com.att.dmf.mr.utils.DMaaPResponseBuilder;
+import com.att.dmf.mr.utils.Utils;
 import com.att.nsa.configs.ConfigDbException;
 import com.att.nsa.dmaap.mmagent.*;
 import com.att.nsa.drumlin.till.nv.rrNvReadable.missingReqdSetting;
@@ -55,19 +55,19 @@ import com.google.gson.JsonSyntaxException;
 import edu.emory.mathcs.backport.java.util.Arrays;
 
 import com.att.ajsc.filemonitor.AJSCPropertiesMap;
-import com.att.nsa.cambria.CambriaApiException;
-import com.att.nsa.cambria.backends.ConsumerFactory.UnavailableException;
+import com.att.dmf.mr.CambriaApiException;
+import com.att.dmf.mr.backends.ConsumerFactory.UnavailableException;
 
 import org.json.JSONArray;
 import org.json.JSONException;
-import com.att.nsa.cambria.beans.DMaaPContext;
-import com.att.nsa.cambria.constants.CambriaConstants;
-import com.att.nsa.cambria.exception.DMaaPErrorMessages;
-import com.att.nsa.cambria.metabroker.Broker.TopicExistsException;
-import com.att.nsa.cambria.security.DMaaPAAFAuthenticator;
-import com.att.nsa.cambria.security.DMaaPAAFAuthenticatorImpl;
-import com.att.nsa.cambria.security.DMaaPAuthenticatorImpl;
-import com.att.nsa.cambria.service.MMService;
+import com.att.dmf.mr.beans.DMaaPContext;
+import com.att.dmf.mr.constants.CambriaConstants;
+import com.att.dmf.mr.exception.DMaaPErrorMessages;
+import com.att.dmf.mr.metabroker.Broker.TopicExistsException;
+import com.att.dmf.mr.security.DMaaPAAFAuthenticator;
+import com.att.dmf.mr.security.DMaaPAAFAuthenticatorImpl;
+import com.att.dmf.mr.security.DMaaPAuthenticatorImpl;
+import com.att.dmf.mr.service.MMService;
 
 /**
  * Rest Service class for Mirror Maker proxy Rest Services
diff --git a/src/main/java/com/att/nsa/dmaap/service/MetricsRestService.java b/src/main/java/com/att/nsa/dmaap/service/MetricsRestService.java
index 8a6240e..2b7b560 100644
--- a/src/main/java/com/att/nsa/dmaap/service/MetricsRestService.java
+++ b/src/main/java/com/att/nsa/dmaap/service/MetricsRestService.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -38,17 +38,17 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Component;
 
-import com.att.nsa.cambria.CambriaApiException;
-import com.att.nsa.cambria.beans.DMaaPContext;
-import com.att.nsa.cambria.exception.DMaaPResponseCode;
-import com.att.nsa.cambria.exception.ErrorResponse;
-import com.att.nsa.cambria.service.MetricsService;
-import com.att.nsa.cambria.utils.ConfigurationReader;
+import com.att.dmf.mr.CambriaApiException;
+import com.att.dmf.mr.beans.DMaaPContext;
+import com.att.dmf.mr.exception.DMaaPResponseCode;
+import com.att.dmf.mr.exception.ErrorResponse;
+import com.att.dmf.mr.service.MetricsService;
+import com.att.dmf.mr.utils.ConfigurationReader;
 
 /**
  * This class is a CXF REST service which acts 
  * as gateway for MR Metrics Service.
- * @author author
+ * @author rajashree.khare
  *
  */
 @Component
diff --git a/src/main/java/com/att/nsa/dmaap/service/ServiceUtil.java b/src/main/java/com/att/nsa/dmaap/service/ServiceUtil.java
index 928ab9f..4045ae3 100644
--- a/src/main/java/com/att/nsa/dmaap/service/ServiceUtil.java
+++ b/src/main/java/com/att/nsa/dmaap/service/ServiceUtil.java
@@ -19,10 +19,10 @@
  */
 package com.att.nsa.dmaap.service;
 
-import com.att.nsa.cambria.beans.DMaaPContext;
+import com.att.dmf.mr.beans.DMaaPContext;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
-import com.att.nsa.cambria.utils.ConfigurationReader;
+import com.att.dmf.mr.utils.ConfigurationReader;
 
 public class ServiceUtil {
 	private static DMaaPContext dmaaPContext;
diff --git a/src/main/java/com/att/nsa/dmaap/service/TopicRestService.java b/src/main/java/com/att/nsa/dmaap/service/TopicRestService.java
index d8be745..3540664 100644
--- a/src/main/java/com/att/nsa/dmaap/service/TopicRestService.java
+++ b/src/main/java/com/att/nsa/dmaap/service/TopicRestService.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -45,26 +45,26 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
-import com.att.nsa.cambria.CambriaApiException;
-import com.att.nsa.cambria.beans.DMaaPContext;
-import com.att.nsa.cambria.beans.TopicBean;
-import com.att.nsa.cambria.constants.CambriaConstants;
-import com.att.nsa.cambria.exception.DMaaPAccessDeniedException;
-import com.att.nsa.cambria.exception.DMaaPErrorMessages;
-import com.att.nsa.cambria.exception.DMaaPResponseCode;
-import com.att.nsa.cambria.exception.ErrorResponse;
-import com.att.nsa.cambria.metabroker.Broker.TopicExistsException;
-import com.att.nsa.cambria.security.DMaaPAAFAuthenticator;
-import com.att.nsa.cambria.security.DMaaPAAFAuthenticatorImpl;
-import com.att.nsa.cambria.service.TopicService;
-import com.att.nsa.cambria.utils.ConfigurationReader;
+import com.att.dmf.mr.CambriaApiException;
+import com.att.dmf.mr.beans.DMaaPContext;
+import com.att.dmf.mr.beans.TopicBean;
+import com.att.dmf.mr.constants.CambriaConstants;
+import com.att.dmf.mr.exception.DMaaPAccessDeniedException;
+import com.att.dmf.mr.exception.DMaaPErrorMessages;
+import com.att.dmf.mr.exception.DMaaPResponseCode;
+import com.att.dmf.mr.exception.ErrorResponse;
+import com.att.dmf.mr.metabroker.Broker.TopicExistsException;
+import com.att.dmf.mr.security.DMaaPAAFAuthenticator;
+import com.att.dmf.mr.security.DMaaPAAFAuthenticatorImpl;
+import com.att.dmf.mr.service.TopicService;
+import com.att.dmf.mr.utils.ConfigurationReader;
 import com.att.nsa.configs.ConfigDbException;
 import com.att.nsa.security.ReadWriteSecuredResource.AccessDeniedException;
 
 /**
- * This class is a CXF REST service which acts as gateway for MR Topic Service.
- * 
- * @author author
+ * This class is a CXF REST service which acts 
+ * as gateway for MR Topic Service.
+ * @author Ramkumar Sembaiyan
  *
  */
 
@@ -75,8 +75,7 @@ public class TopicRestService {
 	/**
 	 * Logger obj
 	 */
-	// private static final Logger LOGGER = Logger
-	// .getLogger(TopicRestService.class);
+	//private static final Logger LOGGER = Logger		.getLogger(TopicRestService.class);
 	private static final EELFLogger LOGGER = EELFManager.getInstance().getLogger(TopicRestService.class);
 	/**
 	 * Config Reader
@@ -101,72 +100,76 @@ public class TopicRestService {
 	 * TopicService obj
 	 */
 	@Autowired
-	private TopicService tService;
-
+	private TopicService topicService;
+	
 	/**
 	 * DMaaPErrorMessages obj
 	 */
 	@Autowired
 	private DMaaPErrorMessages errorMessages;
-
-	private DMaaPContext dmaapContext = new DMaaPContext();
-
+	
 	/**
 	 * mrNamespace
 	 */
-	// @Value("${msgRtr.namespace.aaf}")
-	// private String mrNamespace;
+	//@Value("${msgRtr.namespace.aaf}")
+//	private String mrNamespace;
+	
 
 	/**
 	 * Fetches a list of topics from the current kafka instance and converted
 	 * into json object.
 	 * 
 	 * @return list of the topics in json format
-	 * @throws AccessDeniedException
-	 * @throws CambriaApiException
+	 * @throws AccessDeniedException 
+	 * @throws CambriaApiException 
 	 * @throws IOException
 	 * @throws JSONException
-	 */
+	 * */
 	@GET
-	// @Produces(MediaType.TEXT_PLAIN)
+	//@Produces(MediaType.TEXT_PLAIN)
 	public void getTopics() throws CambriaApiException {
 		try {
-
+						
 			LOGGER.info("Authenticating the user before fetching the topics");
-			// String permission = "com.att.dmaap.mr.topic|*|view";
-			String mrNameS = com.att.ajsc.beans.PropertiesMapBean.getProperty(CambriaConstants.msgRtr_prop,
-					"msgRtr.namespace.aaf");
-			String permission = mrNameS + "|" + "*" + "|" + "view";
+			//String permission = "com.att.dmaap.mr.topic|*|view";
+			String mrNameS= com.att.ajsc.beans.PropertiesMapBean.getProperty(CambriaConstants.msgRtr_prop,"msgRtr.namespace.aaf");
+			String permission =mrNameS+"|"+"*"+"|"+"view";
 			DMaaPAAFAuthenticator aaf = new DMaaPAAFAuthenticatorImpl();
-			// Check if client is using AAF CADI Basic Authorization
-			// If yes then check for AAF role authentication else display all
-			// topics
-			if (null != getDmaapContext().getRequest().getHeader("Authorization")) {
-				if (!aaf.aafAuthentication(getDmaapContext().getRequest(), permission)) {
-
-					ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN,
-							DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(),
-							errorMessages.getNotPermitted1() + " read " + errorMessages.getNotPermitted2());
+			//Check if client is using AAF CADI Basic Authorization
+			//If yes then check for AAF role authentication else display all topics 
+			if(null!=getDmaapContext().getRequest().getHeader("Authorization"))
+			{
+				if(!aaf.aafAuthentication(getDmaapContext().getRequest(), permission))
+				{
+				
+					ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN, 
+							DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(), 
+							errorMessages.getNotPermitted1()+" read "+errorMessages.getNotPermitted2());
 					LOGGER.info(errRes.toString());
 					throw new DMaaPAccessDeniedException(errRes);
-
+					
+				
 				}
-			}
-
-			LOGGER.info("Fetching all Topics");
-
-			tService.getTopics(getDmaapContext());
-
-			LOGGER.info("Returning List of all Topics");
-
+			}	
+			
+				LOGGER.info("Fetching all Topics");
+					//topicService = new com.att.dmf.mr.service.impl.TopicServiceImpl();
+				topicService.getTopics(getDmaapContext());
+	
+				LOGGER.info("Returning List of all Topics");
+			
+			
 		} catch (JSONException | ConfigDbException | IOException excp) {
-			LOGGER.error("Failed to retrieve list of all topics: " + excp.getMessage(), excp);
-
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND,
-					DMaaPResponseCode.GET_TOPICS_FAIL.getResponseCode(),
-					errorMessages.getTopicsfailure() + excp.getMessage());
+			LOGGER.error(
+					"Failed to retrieve list of all topics: "
+							+ excp.getMessage(), excp);
+			
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND, 
+					DMaaPResponseCode.GET_TOPICS_FAIL.getResponseCode(), 
+					errorMessages.getTopicsfailure()+ excp.getMessage());
 			LOGGER.info(errRes.toString());
 			throw new CambriaApiException(errRes);
+			
 
 		}
 	}
@@ -176,56 +179,62 @@ public class TopicRestService {
 	 * into json object.
 	 * 
 	 * @return list of the topics in json format
-	 * @throws AccessDeniedException
-	 * @throws CambriaApiException
+	 * @throws AccessDeniedException 
+	 * @throws CambriaApiException 
 	 * @throws IOException
 	 * @throws JSONException
-	 */
+	 * */
 	@GET
 	@Path("/listAll")
-	// @Produces(MediaType.TEXT_PLAIN)
+	//@Produces(MediaType.TEXT_PLAIN)
 	public void getAllTopics() throws CambriaApiException {
 		try {
-
+						
 			LOGGER.info("Authenticating the user before fetching the topics");
-			// String permission = "com.att.dmaap.mr.topic|*|view";
-			String mrNameS = com.att.ajsc.beans.PropertiesMapBean.getProperty(CambriaConstants.msgRtr_prop,
-					"msgRtr.namespace.aaf");
-			String permission = mrNameS + "|" + "*" + "|" + "view";
+			//String permission = "com.att.dmaap.mr.topic|*|view";
+			String mrNameS= com.att.ajsc.beans.PropertiesMapBean.getProperty(CambriaConstants.msgRtr_prop,"msgRtr.namespace.aaf");
+			String permission =mrNameS+"|"+"*"+"|"+"view";
 			DMaaPAAFAuthenticator aaf = new DMaaPAAFAuthenticatorImpl();
-			// Check if client is using AAF CADI Basic Authorization
-			// If yes then check for AAF role authentication else display all
-			// topics
-			if (null != getDmaapContext().getRequest().getHeader("Authorization")) {
-				if (!aaf.aafAuthentication(getDmaapContext().getRequest(), permission)) {
-
-					ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN,
-							DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(),
-							errorMessages.getNotPermitted1() + " read " + errorMessages.getNotPermitted2());
+			//Check if client is using AAF CADI Basic Authorization
+			//If yes then check for AAF role authentication else display all topics 
+			if(null!=getDmaapContext().getRequest().getHeader("Authorization"))
+			{
+				if(!aaf.aafAuthentication(getDmaapContext().getRequest(), permission))
+				{
+				
+					ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN, 
+							DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(), 
+							errorMessages.getNotPermitted1()+" read "+errorMessages.getNotPermitted2());
 					LOGGER.info(errRes.toString());
 					throw new DMaaPAccessDeniedException(errRes);
-
+					
+				
 				}
-			}
-
-			LOGGER.info("Fetching all Topics");
-
-			tService.getAllTopics(getDmaapContext());
-
-			LOGGER.info("Returning List of all Topics");
-
+			}	
+			
+				LOGGER.info("Fetching all Topics");
+				
+				topicService.getAllTopics(getDmaapContext());
+	
+				LOGGER.info("Returning List of all Topics");
+			
+			
 		} catch (JSONException | ConfigDbException | IOException excp) {
-			LOGGER.error("Failed to retrieve list of all topics: " + excp.getMessage(), excp);
-
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND,
-					DMaaPResponseCode.GET_TOPICS_FAIL.getResponseCode(),
-					errorMessages.getTopicsfailure() + excp.getMessage());
+			LOGGER.error(
+					"Failed to retrieve list of all topics: "
+							+ excp.getMessage(), excp);
+			
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND, 
+					DMaaPResponseCode.GET_TOPICS_FAIL.getResponseCode(), 
+					errorMessages.getTopicsfailure()+ excp.getMessage());
 			LOGGER.info(errRes.toString());
 			throw new CambriaApiException(errRes);
+			
 
 		}
 	}
 
+	
 	/**
 	 * Returns details of the topic whose name is passed as a parameter
 	 * 
@@ -233,55 +242,59 @@ public class TopicRestService {
 	 *            - name of the topic
 	 * @return details of a topic whose name is mentioned in the request in json
 	 *         format.
-	 * @throws AccessDeniedException
-	 * @throws DMaaPAccessDeniedException
+	 * @throws AccessDeniedException 
+	 * @throws DMaaPAccessDeniedException 
 	 * @throws IOException
-	 */
+	 * */
 	@GET
 	@Path("/{topicName}")
-	// @Produces(MediaType.TEXT_PLAIN)
+	//@Produces(MediaType.TEXT_PLAIN)
 	public void getTopic(@PathParam("topicName") String topicName) throws CambriaApiException {
 		try {
-
-			LOGGER.info("Authenticating the user before fetching the details about topic = " + topicName);
+			
+			LOGGER.info("Authenticating the user before fetching the details about topic = "+ topicName);
 			DMaaPAAFAuthenticator aaf = new DMaaPAAFAuthenticatorImpl();
-
-			// String permission=
-			// "com.att.ecomp_test.crm.mr.topic|:topic.com.att.ecomp_test.crm.preDemo|view";
-
-			// Check if client is using AAF CADI Basic Authorization
-			// If yes then check for AAF role authentication else display all
-			// topics
-			if (null != getDmaapContext().getRequest().getHeader("Authorization")) {
+			
+			//String permission= "com.att.ecomp_test.crm.mr.topic|:topic.com.att.ecomp_test.crm.preDemo|view";
+			
+			//Check if client is using AAF CADI Basic Authorization
+			//If yes then check for AAF role authentication else display all topics 
+			if(null!=getDmaapContext().getRequest().getHeader("Authorization"))
+			{
 				String permission = aaf.aafPermissionString(topicName, "view");
-				if (!aaf.aafAuthentication(getDmaapContext().getRequest(), permission)) {
-
-					ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN,
-							DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(),
-							errorMessages.getNotPermitted1() + " read " + errorMessages.getNotPermitted2());
-					LOGGER.info(errRes.toString());
-					throw new DMaaPAccessDeniedException(errRes);
-				}
-			}
-
-			LOGGER.info("Fetching Topic: " + topicName);
-
-			tService.getTopic(getDmaapContext(), topicName);
-
-			LOGGER.info("Fetched details of topic: " + topicName);
-
+				if(!aaf.aafAuthentication(getDmaapContext().getRequest(), permission))
+				{
+						
+						ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN, 
+								DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(), 
+								errorMessages.getNotPermitted1()+" read "+errorMessages.getNotPermitted2());
+						LOGGER.info(errRes.toString());
+						throw new DMaaPAccessDeniedException(errRes);
+					}
+			} 
+			
+				LOGGER.info("Fetching Topic: " + topicName);
+				
+				topicService.getTopic(getDmaapContext(), topicName);
+
+				LOGGER.info("Fetched details of topic: " + topicName);
+		
 		} catch (ConfigDbException | IOException | TopicExistsException excp) {
-			LOGGER.error("Failed to retrieve details of topic: " + topicName, excp);
-
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND,
-					DMaaPResponseCode.GET_TOPICS_DETAILS_FAIL.getResponseCode(),
-					errorMessages.getTopicDetailsFail() + topicName + excp.getMessage());
+			LOGGER.error("Failed to retrieve details of topic: " + topicName,
+					excp);
+			
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND, 
+					DMaaPResponseCode.GET_TOPICS_DETAILS_FAIL.getResponseCode(), 
+					errorMessages.getTopicDetailsFail()+topicName+ excp.getMessage());
 			LOGGER.info(errRes.toString());
 			throw new CambriaApiException(errRes);
-
+			
+			
 		}
 	}
 
+	
+	
 	/**
 	 * This method is still not working. Need to check on post call and how to
 	 * accept parameters for post call
@@ -290,50 +303,57 @@ public class TopicRestService {
 	 *            it will have the bean object
 	 * @throws TopicExistsException
 	 * @throws CambriaApiException
-	 * @throws JSONException
+	 * @throws JSONException 
 	 * @throws IOException
 	 * @throws AccessDeniedException
 	 * 
-	 */
+	 * */
 	@POST
 	@Path("/create")
 	@Consumes({ MediaType.APPLICATION_JSON })
-	// @Produces(MediaType.TEXT_PLAIN)
-	public void createTopic(TopicBean topicBean) throws CambriaApiException{
-		try {
-			LOGGER.info("Creating Topic." + topicBean.getTopicName());
-
-			tService.createTopic(getDmaapContext(), topicBean);
+	//@Produces(MediaType.TEXT_PLAIN)
+	public void createTopic(TopicBean topicBean) throws CambriaApiException, JSONException {
+	try {
+			LOGGER.info("Creating Topic."+topicBean.getTopicName());
+		
+			topicService.createTopic(getDmaapContext(), topicBean);
 
 			LOGGER.info("Topic created Successfully.");
-		} catch (TopicExistsException ex) {
-
-			LOGGER.error("Error while creating a topic: " + ex.getMessage(), ex);
-
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_CONFLICT,
-					DMaaPResponseCode.CREATE_TOPIC_FAIL.getResponseCode(),
-					errorMessages.getCreateTopicFail() + ex.getMessage());
-			LOGGER.info(errRes.toString());
-			throw new CambriaApiException(errRes);
-
-		} catch (AccessDeniedException | DMaaPAccessDeniedException excp) {
-			LOGGER.error("Error while creating a topic: " + excp.getMessage(), excp);
-
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_UNAUTHORIZED,
-					DMaaPResponseCode.CREATE_TOPIC_FAIL.getResponseCode(),
-					errorMessages.getCreateTopicFail() + excp.getMessage());
+		} 
+	catch (TopicExistsException ex){
+		
+		LOGGER.error("Error while creating a topic: " + ex.getMessage(),
+				ex);
+		
+		ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_CONFLICT, 
+				DMaaPResponseCode.CREATE_TOPIC_FAIL.getResponseCode(), 
+				errorMessages.getCreateTopicFail()+ ex.getMessage());
+		LOGGER.info(errRes.toString());
+		throw new CambriaApiException(errRes);
+		
+		
+	
+	
+		}catch (AccessDeniedException | DMaaPAccessDeniedException excp) {
+			LOGGER.error("Error while creating a topic: " + excp.getMessage(),
+					excp);
+			
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_UNAUTHORIZED, 
+					DMaaPResponseCode.CREATE_TOPIC_FAIL.getResponseCode(), 
+					errorMessages.getCreateTopicFail()+ excp.getMessage());
 			LOGGER.info(errRes.toString());
 			throw new CambriaApiException(errRes);
-
-		} catch (CambriaApiException | IOException excp) {
-			LOGGER.error("Error while creating a topic: " + excp.getMessage(), excp);
-
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND,
-					DMaaPResponseCode.CREATE_TOPIC_FAIL.getResponseCode(),
-					errorMessages.getCreateTopicFail() + excp.getMessage());
+			
+		}catch (CambriaApiException |  IOException			 excp) {
+			LOGGER.error("Error while creating a topic: " + excp.getMessage(),
+					excp);
+			
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND, 
+					DMaaPResponseCode.CREATE_TOPIC_FAIL.getResponseCode(), 
+					errorMessages.getCreateTopicFail()+ excp.getMessage());
 			LOGGER.info(errRes.toString());
 			throw new CambriaApiException(errRes);
-
+			
 		}
 	}
 
@@ -342,42 +362,45 @@ public class TopicRestService {
 	 * 
 	 * @param topicName
 	 *            topic
-	 * @throws CambriaApiException
+	 * @throws CambriaApiException 
 	 * @throws IOException
-	 */
+	 * */
 	@DELETE
 	@Path("/{topicName}")
-	// @Produces(MediaType.TEXT_PLAIN)
+	//@Produces(MediaType.TEXT_PLAIN)
 	public void deleteTopic(@PathParam("topicName") String topicName) throws CambriaApiException {
 		try {
 			LOGGER.info("Deleting Topic: " + topicName);
 
-			tService.deleteTopic(getDmaapContext(), topicName);
+			topicService.deleteTopic(getDmaapContext(), topicName);
 
 			LOGGER.info("Topic [" + topicName + "] deleted successfully.");
-		} catch (DMaaPAccessDeniedException | AccessDeniedException excp) {
-			LOGGER.error("Error while creating a topic: " + excp.getMessage(), excp);
-
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_UNAUTHORIZED,
-					DMaaPResponseCode.CREATE_TOPIC_FAIL.getResponseCode(),
-					errorMessages.getCreateTopicFail() + excp.getMessage());
+		} catch (DMaaPAccessDeniedException| AccessDeniedException excp) {
+			LOGGER.error("Error while creating a topic: " + excp.getMessage(),
+					excp);
+			
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_UNAUTHORIZED, 
+					DMaaPResponseCode.CREATE_TOPIC_FAIL.getResponseCode(), 
+					errorMessages.getCreateTopicFail()+ excp.getMessage());
 			LOGGER.info(errRes.toString());
 			throw new CambriaApiException(errRes);
-
-		} catch (IOException | ConfigDbException | CambriaApiException | TopicExistsException excp) {
+			
+	}catch (IOException | ConfigDbException
+				| CambriaApiException | TopicExistsException excp) {
 			LOGGER.error("Error while deleting topic: " + topicName, excp);
-
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND,
-					DMaaPResponseCode.DELETE_TOPIC_FAIL.getResponseCode(),
-					errorMessages.getDeleteTopicFail() + topicName + excp.getMessage());
+		
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND, 
+					DMaaPResponseCode.DELETE_TOPIC_FAIL.getResponseCode(), 
+					errorMessages.getDeleteTopicFail()+ topicName + excp.getMessage());
 			LOGGER.info(errRes.toString());
 			throw new CambriaApiException(errRes);
-
+			
 		}
 	}
 
 	private DMaaPContext getDmaapContext() {
 
+		DMaaPContext dmaapContext = new DMaaPContext();
 		dmaapContext.setRequest(request);
 		dmaapContext.setResponse(response);
 		dmaapContext.setConfigReader(configReader);
@@ -390,48 +413,49 @@ public class TopicRestService {
 	 * This method will fetch the details of publisher by giving topic name
 	 * 
 	 * @param topicName
-	 * @throws CambriaApiException
-	 * @throws AccessDeniedException
+	 * @throws CambriaApiException 
+	 * @throws AccessDeniedException 
 	 */
 	@GET
 	@Path("/{topicName}/producers")
-	// @Produces(MediaType.TEXT_PLAIN)
-	public void getPublishersByTopicName(@PathParam("topicName") String topicName) throws CambriaApiException {
+	//@Produces(MediaType.TEXT_PLAIN)
+	public void getPublishersByTopicName(
+			@PathParam("topicName") String topicName) throws CambriaApiException {
 		try {
-
-			// String permission =
-			// "com.att.dmaap.mr.topic"+"|"+topicName+"|"+"manage";
-			// DMaaPAAFAuthenticator aaf = new DMaaPAAFAuthenticatorImpl();
-			// String permission = aaf.aafPermissionString(topicName, "view");
-			// if(aaf.aafAuthentication(getDmaapContext().getRequest(),
-			// permission))
-			// {
-			LOGGER.info("Fetching list of all the publishers for topic " + topicName);
-
-			tService.getPublishersByTopicName(getDmaapContext(), topicName);
-
-			LOGGER.info("Returning list of all the publishers for topic " + topicName);
-			// }else{
-			// LOGGER.error("Error while fetching list of publishers for topic
-			// "+ topicName);
-			//
-			// ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN,
-			// DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(),
-			// errorMessages.getNotPermitted1()+" fetch list of publishers
-			// "+errorMessages.getNotPermitted2());
-			// LOGGER.info(errRes);
-			// throw new DMaaPAccessDeniedException(errRes);
-			//
-			// }
-
+			
+//			String permission = "com.att.dmaap.mr.topic"+"|"+topicName+"|"+"manage";
+//			DMaaPAAFAuthenticator aaf = new DMaaPAAFAuthenticatorImpl();
+//			String permission = aaf.aafPermissionString(topicName, "view");
+//			if(aaf.aafAuthentication(getDmaapContext().getRequest(), permission))
+//			{
+				LOGGER.info("Fetching list of all the publishers for topic "
+						+ topicName);
+
+				topicService.getPublishersByTopicName(getDmaapContext(), topicName);
+
+				LOGGER.info("Returning list of all the publishers for topic "
+						+ topicName);
+//			}else{
+//				LOGGER.error("Error while fetching list of publishers for topic "+ topicName);
+//				
+//				ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN, 
+//						DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(), 
+//						errorMessages.getNotPermitted1()+" fetch list of publishers "+errorMessages.getNotPermitted2());
+//				LOGGER.info(errRes);
+//				throw new DMaaPAccessDeniedException(errRes);
+//				
+//			}
+			
 		} catch (IOException | ConfigDbException | TopicExistsException excp) {
-			LOGGER.error("Error while fetching list of publishers for topic " + topicName, excp);
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND,
-					DMaaPResponseCode.GET_PUBLISHERS_BY_TOPIC.getResponseCode(),
-					"Error while fetching list of publishers for topic: " + topicName + excp.getMessage());
+			LOGGER.error("Error while fetching list of publishers for topic "
+					+ topicName, excp);
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND, 
+					DMaaPResponseCode.GET_PUBLISHERS_BY_TOPIC.getResponseCode(), 
+					"Error while fetching list of publishers for topic: "
+							+ topicName + excp.getMessage());
 			LOGGER.info(errRes.toString());
 			throw new CambriaApiException(errRes);
-
+			
 		}
 	}
 
@@ -440,38 +464,44 @@ public class TopicRestService {
 	 * 
 	 * @param topicName
 	 * @param producerId
-	 * @throws CambriaApiException
+	 * @throws CambriaApiException 
 	 */
 	@PUT
 	@Path("/{topicName}/producers/{producerId}")
-	public void permitPublisherForTopic(@PathParam("topicName") String topicName,
+	public void permitPublisherForTopic(
+			@PathParam("topicName") String topicName,
 			@PathParam("producerId") String producerId) throws CambriaApiException {
 		try {
-			LOGGER.info("Granting write access to producer [" + producerId + "] for topic " + topicName);
+			LOGGER.info("Granting write access to producer [" + producerId
+					+ "] for topic " + topicName);
 
-			tService.permitPublisherForTopic(getDmaapContext(), topicName, producerId);
+			topicService.permitPublisherForTopic(getDmaapContext(), topicName,
+					producerId);
 
-			LOGGER.info("Write access has been granted to producer [" + producerId + "] for topic " + topicName);
+			LOGGER.info("Write access has been granted to producer ["
+					+ producerId + "] for topic " + topicName);
 		} catch (AccessDeniedException | DMaaPAccessDeniedException excp) {
-			LOGGER.error("Error while creating a topic: " + excp.getMessage(), excp);
-
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_UNAUTHORIZED,
-					DMaaPResponseCode.CREATE_TOPIC_FAIL.getResponseCode(),
-					errorMessages.getCreateTopicFail() + excp.getMessage());
+			LOGGER.error("Error while creating a topic: " + excp.getMessage(),
+					excp);
+			
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_UNAUTHORIZED, 
+					DMaaPResponseCode.CREATE_TOPIC_FAIL.getResponseCode(), 
+					errorMessages.getCreateTopicFail()+ excp.getMessage());
 			LOGGER.info(errRes.toString());
 			throw new CambriaApiException(errRes);
-
-		} catch (ConfigDbException | IOException | TopicExistsException excp) {
-			LOGGER.error("Error while granting write access to producer [" + producerId + "] for topic " + topicName,
-					excp);
-
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND,
-					DMaaPResponseCode.PERMIT_PUBLISHER_FOR_TOPIC.getResponseCode(),
-					"Error while granting write access to producer [" + producerId + "] for topic " + topicName
-							+ excp.getMessage());
+			
+	  }catch ( ConfigDbException | IOException
+				| TopicExistsException excp) {
+			LOGGER.error("Error while granting write access to producer ["
+					+ producerId + "] for topic " + topicName, excp);
+			
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND, 
+					DMaaPResponseCode.PERMIT_PUBLISHER_FOR_TOPIC.getResponseCode(), 
+					"Error while granting write access to producer ["
+							+ producerId + "] for topic " + topicName + excp.getMessage());
 			LOGGER.info(errRes.toString());
 			throw new CambriaApiException(errRes);
-
+			
 		}
 	}
 
@@ -480,34 +510,39 @@ public class TopicRestService {
 	 * 
 	 * @param topicName
 	 * @param producerId
-	 * @throws CambriaApiException
+	 * @throws CambriaApiException 
 	 */
 	@DELETE
 	@Path("/{topicName}/producers/{producerId}")
 	public void denyPublisherForTopic(@PathParam("topicName") String topicName,
 			@PathParam("producerId") String producerId) throws CambriaApiException {
 		try {
-			LOGGER.info("Revoking write access to producer [" + producerId + "] for topic " + topicName);
+			LOGGER.info("Revoking write access to producer [" + producerId
+					+ "] for topic " + topicName);
 
-			tService.denyPublisherForTopic(getDmaapContext(), topicName, producerId);
+			topicService.denyPublisherForTopic(getDmaapContext(), topicName,
+					producerId);
 
-			LOGGER.info("Write access revoked for producer [" + producerId + "] for topic " + topicName);
+			LOGGER.info("Write access revoked for producer [" + producerId
+					+ "] for topic " + topicName);
 		} catch (DMaaPAccessDeniedException | AccessDeniedException excp) {
-			LOGGER.error("Error while creating a topic: " + excp.getMessage(), excp);
-
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_UNAUTHORIZED,
-					DMaaPResponseCode.CREATE_TOPIC_FAIL.getResponseCode(),
-					errorMessages.getCreateTopicFail() + excp.getMessage());
+			LOGGER.error("Error while creating a topic: " + excp.getMessage(),
+					excp);
+			
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_UNAUTHORIZED, 
+					DMaaPResponseCode.CREATE_TOPIC_FAIL.getResponseCode(), 
+					errorMessages.getCreateTopicFail()+ excp.getMessage());
 			LOGGER.info(errRes.toString());
 			throw new CambriaApiException(errRes);
-
-		} catch (ConfigDbException | IOException | TopicExistsException excp) {
-			LOGGER.error("Error while revoking write access for producer [" + producerId + "] for topic " + topicName,
-					excp);
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN,
-					DMaaPResponseCode.REVOKE_PUBLISHER_FOR_TOPIC.getResponseCode(),
-					"Error while revoking write access to producer [" + producerId + "] for topic " + topicName
-							+ excp.getMessage());
+			
+	}catch ( ConfigDbException | IOException
+				| TopicExistsException excp) {
+			LOGGER.error("Error while revoking write access for producer ["
+					+ producerId + "] for topic " + topicName, excp);
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN, 
+					DMaaPResponseCode.REVOKE_PUBLISHER_FOR_TOPIC.getResponseCode(), 
+					"Error while revoking write access to producer ["
+							+ producerId + "] for topic " + topicName + excp.getMessage());
 			LOGGER.info(errRes.toString());
 			throw new CambriaApiException(errRes);
 		}
@@ -517,51 +552,55 @@ public class TopicRestService {
 	 * Get the consumer details by the topic name
 	 * 
 	 * @param topicName
-	 * @throws AccessDeniedException
-	 * @throws CambriaApiException
+	 * @throws AccessDeniedException 
+	 * @throws CambriaApiException 
 	 */
 	@GET
 	@Path("/{topicName}/consumers")
-	// @Produces(MediaType.TEXT_PLAIN)
-	public void getConsumersByTopicName(@PathParam("topicName") String topicName)
-			throws CambriaApiException {
+	//@Produces(MediaType.TEXT_PLAIN)
+	public void getConsumersByTopicName(@PathParam("topicName") String topicName) throws AccessDeniedException, 
+	CambriaApiException {
 		try {
-
-			// String permission =
-			// "com.att.dmaap.mr.topic"+"|"+topicName+"|"+"view";
-			// DMaaPAAFAuthenticator aaf = new DMaaPAAFAuthenticatorImpl();
-			// String permission = aaf.aafPermissionString(topicName, "view");
-			// if(aaf.aafAuthentication(getDmaapContext().getRequest(),
-			// permission))
-			// {
-			LOGGER.info("Fetching list of all consumers for topic " + topicName);
-
-			tService.getConsumersByTopicName(getDmaapContext(), topicName);
-
-			LOGGER.info("Returning list of all consumers for topic " + topicName);
-
-			// }else{
-			// LOGGER.error(
-			// "Error while fetching list of all consumers for topic "
-			// + topicName);
-			// ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN,
-			// DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(),
-			// errorMessages.getNotPermitted1()+" fetch list of consumers
-			// "+errorMessages.getNotPermitted2());
-			// LOGGER.info(errRes);
-			// throw new DMaaPAccessDeniedException(errRes);
-			//
-			//
-			// }
-
+			
+			
+//			String permission = "com.att.dmaap.mr.topic"+"|"+topicName+"|"+"view";
+//			DMaaPAAFAuthenticator aaf = new DMaaPAAFAuthenticatorImpl();
+//			String permission = aaf.aafPermissionString(topicName, "view");
+//			if(aaf.aafAuthentication(getDmaapContext().getRequest(), permission))
+//			{
+				LOGGER.info("Fetching list of all consumers for topic " + topicName);
+
+				topicService.getConsumersByTopicName(getDmaapContext(), topicName);
+
+				LOGGER.info("Returning list of all consumers for topic "
+						+ topicName);
+				
+//			}else{
+//				LOGGER.error(
+//						"Error while fetching list of all consumers for topic "
+//								+ topicName);
+//				ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN, 
+//						DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(), 
+//						errorMessages.getNotPermitted1()+" fetch list of consumers "+errorMessages.getNotPermitted2());
+//				LOGGER.info(errRes);
+//				throw new DMaaPAccessDeniedException(errRes);
+//				
+//				
+//			}
+			
+			
+			
 		} catch (IOException | ConfigDbException | TopicExistsException excp) {
-			LOGGER.error("Error while fetching list of all consumers for topic " + topicName, excp);
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN,
-					DMaaPResponseCode.GET_CONSUMERS_BY_TOPIC.getResponseCode(),
-					"Error while fetching list of all consumers for topic: " + topicName + excp.getMessage());
+			LOGGER.error(
+					"Error while fetching list of all consumers for topic "
+							+ topicName, excp);
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN, 
+					DMaaPResponseCode.GET_CONSUMERS_BY_TOPIC.getResponseCode(), 
+					"Error while fetching list of all consumers for topic: "
+							+ topicName+ excp.getMessage());
 			LOGGER.info(errRes.toString());
 			throw new CambriaApiException(errRes);
-
+			
 		}
 	}
 
@@ -570,28 +609,33 @@ public class TopicRestService {
 	 * 
 	 * @param topicName
 	 * @param consumerId
-	 * @throws CambriaApiException
+	 * @throws CambriaApiException 
 	 */
 	@PUT
 	@Path("/{topicName}/consumers/{consumerId}")
-	public void permitConsumerForTopic(@PathParam("topicName") String topicName,
+	public void permitConsumerForTopic(
+			@PathParam("topicName") String topicName,
 			@PathParam("consumerId") String consumerId) throws CambriaApiException {
 		try {
-			LOGGER.info("Granting read access to consumer [" + consumerId + "] for topic " + topicName);
-
-			tService.permitConsumerForTopic(getDmaapContext(), topicName, consumerId);
-
-			LOGGER.info("Read access granted to consumer [" + consumerId + "] for topic " + topicName);
-		} catch (AccessDeniedException | ConfigDbException | IOException | TopicExistsException excp) {
-			LOGGER.error("Error while granting read access to consumer [" + consumerId + "] for topic " + topicName,
-					excp);
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN,
-					DMaaPResponseCode.PERMIT_CONSUMER_FOR_TOPIC.getResponseCode(),
-					"Error while granting read access to consumer [" + consumerId + "] for topic " + topicName
-							+ excp.getMessage());
+			LOGGER.info("Granting read access to consumer [" + consumerId
+					+ "] for topic " + topicName);
+
+			topicService.permitConsumerForTopic(getDmaapContext(), topicName,
+					consumerId);
+
+			LOGGER.info("Read access granted to consumer [" + consumerId
+					+ "] for topic " + topicName);
+		} catch (AccessDeniedException | ConfigDbException | IOException
+				| TopicExistsException excp) {
+			LOGGER.error("Error while granting read access to consumer ["
+					+ consumerId + "] for topic " + topicName, excp);
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN, 
+					DMaaPResponseCode.PERMIT_CONSUMER_FOR_TOPIC.getResponseCode(), 
+					"Error while granting read access to consumer ["
+							+ consumerId + "] for topic " + topicName+ excp.getMessage());
 			LOGGER.info(errRes.toString());
 			throw new CambriaApiException(errRes);
-
+			
 		}
 	}
 
@@ -600,37 +644,53 @@ public class TopicRestService {
 	 * 
 	 * @param topicName
 	 * @param consumerId
-	 * @throws CambriaApiException
+	 * @throws CambriaApiException 
 	 */
 	@DELETE
 	@Path("/{topicName}/consumers/{consumerId}")
 	public void denyConsumerForTopic(@PathParam("topicName") String topicName,
 			@PathParam("consumerId") String consumerId) throws CambriaApiException {
 		try {
-			LOGGER.info("Revoking read access to consumer [" + consumerId + "] for topic " + topicName);
-
-			tService.denyConsumerForTopic(getDmaapContext(), topicName, consumerId);
-
-			LOGGER.info("Read access revoked to consumer [" + consumerId + "] for topic " + topicName);
-		} catch (ConfigDbException | IOException | TopicExistsException excp) {
-			LOGGER.error("Error while revoking read access to consumer [" + consumerId + "] for topic " + topicName,
-					excp);
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN,
-					DMaaPResponseCode.REVOKE_CONSUMER_FOR_TOPIC.getResponseCode(),
-					"Error while revoking read access to consumer [" + consumerId + "] for topic " + topicName
-							+ excp.getMessage());
+			LOGGER.info("Revoking read access to consumer [" + consumerId
+					+ "] for topic " + topicName);
+
+			topicService.denyConsumerForTopic(getDmaapContext(), topicName,
+					consumerId);
+
+			LOGGER.info("Read access revoked to consumer [" + consumerId
+					+ "] for topic " + topicName);
+		} catch ( ConfigDbException | IOException
+				| TopicExistsException excp) {
+			LOGGER.error("Error while revoking read access to consumer ["
+					+ consumerId + "] for topic " + topicName, excp);
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN, 
+					DMaaPResponseCode.REVOKE_CONSUMER_FOR_TOPIC.getResponseCode(), 
+					"Error while revoking read access to consumer ["
+							+ consumerId + "] for topic " + topicName+ excp.getMessage());
 			LOGGER.info(errRes.toString());
 			throw new CambriaApiException(errRes);
-		} catch (DMaaPAccessDeniedException | AccessDeniedException excp) {
-			LOGGER.error("Error while creating a topic: " + excp.getMessage(), excp);
-
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_UNAUTHORIZED,
-					DMaaPResponseCode.CREATE_TOPIC_FAIL.getResponseCode(),
-					errorMessages.getCreateTopicFail() + excp.getMessage());
+		}catch (DMaaPAccessDeniedException | AccessDeniedException excp) {
+			LOGGER.error("Error while creating a topic: " + excp.getMessage(),
+					excp);
+			
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_UNAUTHORIZED, 
+					DMaaPResponseCode.CREATE_TOPIC_FAIL.getResponseCode(), 
+					errorMessages.getCreateTopicFail()+ excp.getMessage());
 			LOGGER.info(errRes.toString());
 			throw new CambriaApiException(errRes);
+			
+	}
+	}
 
-		}
+	public TopicService getTopicService() {
+		return topicService;
+	}
+
+	public void setTopicService(TopicService topicService) {
+		this.topicService = topicService;
 	}
 
+
+	
+
 }
diff --git a/src/main/java/com/att/nsa/dmaap/service/TransactionRestService.java b/src/main/java/com/att/nsa/dmaap/service/TransactionRestService.java
index 1a870a1..784f7c5 100644
--- a/src/main/java/com/att/nsa/dmaap/service/TransactionRestService.java
+++ b/src/main/java/com/att/nsa/dmaap/service/TransactionRestService.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -38,19 +38,19 @@ import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Component;
 
 import com.att.aft.dme2.internal.jettison.json.JSONException;
-import com.att.nsa.cambria.CambriaApiException;
-import com.att.nsa.cambria.beans.DMaaPContext;
-import com.att.nsa.cambria.exception.DMaaPResponseCode;
-import com.att.nsa.cambria.exception.ErrorResponse;
-import com.att.nsa.cambria.service.TransactionService;
-import com.att.nsa.cambria.utils.ConfigurationReader;
+import com.att.dmf.mr.CambriaApiException;
+import com.att.dmf.mr.beans.DMaaPContext;
+import com.att.dmf.mr.exception.DMaaPResponseCode;
+import com.att.dmf.mr.exception.ErrorResponse;
+import com.att.dmf.mr.service.TransactionService;
+import com.att.dmf.mr.utils.ConfigurationReader;
 import com.att.nsa.configs.ConfigDbException;
 
 /**
- * This class is a CXF REST service which acts as gateway for DMaaP Transaction
- * Ids.
- * 
- * @author author
+ * This class is a CXF REST service 
+ * which acts as gateway for DMaaP
+ * Transaction Ids.
+ * @author rajashree.khare
  *
  */
 @Component
@@ -84,13 +84,10 @@ public class TransactionRestService {
 	@Autowired
 	private TransactionService transactionService;
 
-	private DMaaPContext dmaapContext = new DMaaPContext();
-
 	/**
 	 * 
 	 * Returns a list of all the existing Transaction Ids
-	 * 
-	 * @throws CambriaApiException
+	 * @throws CambriaApiException 
 	 * 
 	 * @throws IOException
 	 * @exception ConfigDbException
@@ -107,10 +104,11 @@ public class TransactionRestService {
 
 			LOGGER.info("Returning list of all transactions.");
 		} catch (ConfigDbException | IOException e) {
-			LOGGER.error("Error while retrieving list of all transactions: " + e.getMessage(), e);
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_EXPECTATION_FAILED,
-					DMaaPResponseCode.RETRIEVE_TRANSACTIONS.getResponseCode(),
-					"Error while retrieving list of all transactions:" + e.getMessage());
+			LOGGER.error("Error while retrieving list of all transactions: "
+					+ e.getMessage(), e);
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_EXPECTATION_FAILED, 
+					DMaaPResponseCode.RETRIEVE_TRANSACTIONS.getResponseCode(), 
+					"Error while retrieving list of all transactions:"+e.getMessage());
 			LOGGER.info(errRes.toString());
 			throw new CambriaApiException(errRes);
 		}
@@ -123,7 +121,7 @@ public class TransactionRestService {
 	 * 
 	 * @param transactionId
 	 *            - id of transaction
-	 * @throws CambriaApiException
+	 * @throws CambriaApiException 
 	 * @throws IOException
 	 * @exception ConfigDbException
 	 * @exception IOException
@@ -133,18 +131,22 @@ public class TransactionRestService {
 	 */
 	@GET
 	@Path("/{transactionId}")
-	public void getTransactionObj(@PathParam("transactionId") String transactionId) throws CambriaApiException {
+	public void getTransactionObj(
+			@PathParam("transactionId") String transactionId) throws CambriaApiException {
 
 		LOGGER.info("Fetching details of Transaction ID : " + transactionId);
 
 		try {
-			transactionService.getTransactionObj(getDmaapContext(), transactionId);
+			transactionService.getTransactionObj(getDmaapContext(),
+					transactionId);
 		} catch (ConfigDbException | JSONException | IOException e) {
-			LOGGER.error("Error while retrieving transaction details for id: " + transactionId, e);
-
-			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_EXPECTATION_FAILED,
-					DMaaPResponseCode.RETRIEVE_TRANSACTIONS_DETAILS.getResponseCode(),
-					"Error while retrieving transaction details for id: [" + transactionId + "]: " + e.getMessage());
+			LOGGER.error("Error while retrieving transaction details for id: "
+					+ transactionId, e);
+		
+			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_EXPECTATION_FAILED, 
+					DMaaPResponseCode.RETRIEVE_TRANSACTIONS_DETAILS.getResponseCode(), 
+					"Error while retrieving transaction details for id: ["
+							+ transactionId + "]: " + e.getMessage());
 			LOGGER.info(errRes.toString());
 			throw new CambriaApiException(errRes);
 
@@ -163,7 +165,8 @@ public class TransactionRestService {
 	 *         Object,HttpServlet Object
 	 * 
 	 */
-	public DMaaPContext getDmaapContext() {
+	private DMaaPContext getDmaapContext() {
+		DMaaPContext dmaapContext = new DMaaPContext();
 		dmaapContext.setConfigReader(configReader);
 		dmaapContext.setRequest(request);
 		dmaapContext.setResponse(response);
diff --git a/src/main/java/com/att/nsa/dmaap/service/UIRestServices.java b/src/main/java/com/att/nsa/dmaap/service/UIRestServices.java
index 79a39fb..445da5f 100644
--- a/src/main/java/com/att/nsa/dmaap/service/UIRestServices.java
+++ b/src/main/java/com/att/nsa/dmaap/service/UIRestServices.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -30,7 +30,7 @@ import javax.ws.rs.Path;
 import javax.ws.rs.PathParam;
 import javax.ws.rs.core.Context;
 
-import kafka.common.TopicExistsException;
+import org.apache.kafka.common.errors.TopicExistsException;
 
 import org.apache.http.HttpStatus;
 import com.att.eelf.configuration.EELFLogger;
@@ -40,15 +40,15 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Component;
 
-import com.att.nsa.cambria.beans.DMaaPContext;
-import com.att.nsa.cambria.service.UIService;
-import com.att.nsa.cambria.utils.ConfigurationReader;
-import com.att.nsa.cambria.utils.DMaaPResponseBuilder;
+import com.att.dmf.mr.beans.DMaaPContext;
+import com.att.dmf.mr.service.UIService;
+import com.att.dmf.mr.utils.ConfigurationReader;
+import com.att.dmf.mr.utils.DMaaPResponseBuilder;
 import com.att.nsa.configs.ConfigDbException;
 
 /**
  * UI Rest Service
- * @author author
+ * @author rajashree.khare
  *
  */
 @Component
diff --git a/src/main/java/com/att/nsa/dmaap/tools/ConfigTool.java b/src/main/java/com/att/nsa/dmaap/tools/ConfigTool.java
index 9f55249..bf1c1fb 100644
--- a/src/main/java/com/att/nsa/dmaap/tools/ConfigTool.java
+++ b/src/main/java/com/att/nsa/dmaap/tools/ConfigTool.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -31,8 +31,8 @@ import java.util.Map.Entry;
 import org.json.JSONException;
 
 import com.att.nsa.apiServer.CommonServlet;
-import com.att.nsa.cambria.beans.DMaaPKafkaMetaBroker;
-import com.att.nsa.cambria.metabroker.Topic;
+import com.att.dmf.mr.beans.DMaaPKafkaMetaBroker;
+import com.att.dmf.mr.metabroker.Topic;
 import com.att.nsa.cmdtool.Command;
 import com.att.nsa.cmdtool.CommandLineTool;
 import com.att.nsa.cmdtool.CommandNotReadyException;
@@ -128,7 +128,7 @@ public class ConfigTool extends CommandLineTool<ConfigToolContext>
 			}
 			catch ( ConfigDbException e )
 			{
-				out.println ( "Command failed: " + e);
+				out.println ( "Command failed: " + e.getMessage() );
 			}
 		}
 
@@ -167,7 +167,6 @@ public class ConfigTool extends CommandLineTool<ConfigToolContext>
 			catch ( ConfigDbException e )
 			{
 				out.println ( "Command failed: " + e.getMessage() );
-				throw new RuntimeException(e);
 			}
 		}
 
@@ -205,7 +204,7 @@ public class ConfigTool extends CommandLineTool<ConfigToolContext>
 			}
 			catch ( ConfigDbException e )
 			{
-				out.println ( "Command failed: " + e);
+				out.println ( "Command failed: " + e.getMessage() );
 			}
 		}
 
@@ -241,7 +240,7 @@ public class ConfigTool extends CommandLineTool<ConfigToolContext>
 			}
 			catch ( ConfigDbException e )
 			{
-				out.println ( "Command failed: " + e);
+				out.println ( "Command failed: " + e.getMessage () );
 			}
 		}
 
@@ -287,7 +286,7 @@ public class ConfigTool extends CommandLineTool<ConfigToolContext>
 			}
 			catch ( ConfigDbException e )
 			{
-				out.println ( "Command failed: " + e);
+				out.println ( "Command failed: " + e.getMessage () );
 			}
 		}
 
@@ -349,11 +348,11 @@ public class ConfigTool extends CommandLineTool<ConfigToolContext>
 			}
 			catch ( ConfigDbException e )
 			{
-				out.println ( "Command failed: " + e);
+				out.println ( "Command failed: " + e.getMessage() );
 			}
 			catch ( JSONException e )
 			{
-				out.println ( "Command failed: " + e);
+				out.println ( "Command failed: " + e.getMessage() );
 			}
 		}
 
diff --git a/src/main/java/com/att/nsa/dmaap/tools/ConfigToolContext.java b/src/main/java/com/att/nsa/dmaap/tools/ConfigToolContext.java
index bb44d1f..c4369f1 100644
--- a/src/main/java/com/att/nsa/dmaap/tools/ConfigToolContext.java
+++ b/src/main/java/com/att/nsa/dmaap/tools/ConfigToolContext.java
@@ -8,20 +8,20 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
 package com.att.nsa.dmaap.tools;
 
-import com.att.nsa.cambria.beans.DMaaPMetricsSet;
+import com.att.dmf.mr.beans.DMaaPMetricsSet;
 import com.att.nsa.cmdtool.CommandContext;
 import com.att.nsa.configs.ConfigDb;
 import com.att.nsa.drumlin.till.nv.rrNvReadable;
diff --git a/src/main/java/com/att/nsa/dmaap/util/ContentLengthInterceptor.java b/src/main/java/com/att/nsa/dmaap/util/ContentLengthInterceptor.java
index 98bc86e..4ed9322 100644
--- a/src/main/java/com/att/nsa/dmaap/util/ContentLengthInterceptor.java
+++ b/src/main/java/com/att/nsa/dmaap/util/ContentLengthInterceptor.java
@@ -1,24 +1,24 @@
-/*******************************************************************************

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

- *  org.onap.dmaap

- *  ================================================================================

- *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.

- *  ================================================================================

- *  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.

- *  ============LICENSE_END=========================================================

- *

- *  ECOMP is a trademark and service mark of AT&T Intellectual Property.

- *  

- *******************************************************************************/

+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ *  ================================================================================
+ *  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.
+ *  ============LICENSE_END=========================================================
+ *  
+ *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ *  
+ *******************************************************************************/
 package com.att.nsa.dmaap.util;

 

 import java.util.Map;

@@ -30,9 +30,9 @@ import com.att.eelf.configuration.EELFManager;
 import org.json.JSONException;

 import org.json.JSONObject;

 import org.springframework.stereotype.Component;

-import com.att.nsa.cambria.CambriaApiException;

-import com.att.nsa.cambria.exception.DMaaPResponseCode;

-import com.att.nsa.cambria.exception.ErrorResponse;

+import com.att.dmf.mr.CambriaApiException;

+import com.att.dmf.mr.exception.DMaaPResponseCode;

+import com.att.dmf.mr.exception.ErrorResponse;

 import ajsc.beans.interceptors.AjscInterceptor;

 

 /**

@@ -43,6 +43,7 @@ public class ContentLengthInterceptor implements AjscInterceptor{
 

 	

 	private String defLength;

+	//private Logger log = Logger.getLogger(ContentLengthInterceptor.class.toString());

 	private static final EELFLogger log = EELFManager.getInstance().getLogger(ContentLengthInterceptor.class);

 

 

@@ -69,13 +70,13 @@ public class ContentLengthInterceptor implements AjscInterceptor{
 			// checking for no encoding, chunked and requestLength greater then

 			// default length

 				if (null != transferEncoding && !(transferEncoding.contains("chunked"))

-						&& (getDefLength() !=null && requestLength > Integer.parseInt(getDefLength()))) {

+						&& (requestLength > Integer.parseInt(getDefLength()))) {

 					jsonObj = new JSONObject().append("defaultlength", getDefLength())

 							.append("requestlength", requestLength);

 					log.error("message length is greater than default");

 					throw new CambriaApiException(jsonObj);

 				} 

-				else if (null == transferEncoding && (getDefLength() !=null && requestLength > Integer.parseInt(getDefLength()))) 

+				else if (null == transferEncoding && (requestLength > Integer.parseInt(getDefLength()))) 

 				{

 					jsonObj = new JSONObject().append("defaultlength", getDefLength()).append(

 							"requestlength", requestLength);

@@ -93,16 +94,22 @@ public class ContentLengthInterceptor implements AjscInterceptor{
 		} catch (CambriaApiException | NumberFormatException | JSONException e) {

 			

 			log.info("Exception obj--"+e);

-			log.error("message size is greater then default"+e.getMessage());

+			log.error("message size is greater then default"+e.getMessage());
+			String messg=e.toString();
+			if(jsonObj!=null){
+			 messg=jsonObj.toString();
+			}

 			ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_REQUEST_TOO_LONG,

 					DMaaPResponseCode.MSG_SIZE_EXCEEDS_MSG_LIMIT.getResponseCode(), System.getProperty("msg_size_exceeds")

-							+ e.toString());

+							+ messg);

 			log.info(errRes.toString());

 			

 			

 			map.put(HttpServletResponse.SC_INTERNAL_SERVER_ERROR,"test");

-			httpservletresponse.setStatus(HttpStatus.SC_REQUEST_TOO_LONG);

-			httpservletresponse.getOutputStream().write(errRes.toString().getBytes());

+			httpservletresponse.setStatus(HttpStatus.SC_REQUEST_TOO_LONG);
+			if(httpservletresponse.getOutputStream()!=null){

+			httpservletresponse.getOutputStream().write(errRes.toString().getBytes());
+			}

 			return false;

 		}

 

diff --git a/src/main/java/com/att/nsa/dmaap/util/DMaaPAuthFilter.java b/src/main/java/com/att/nsa/dmaap/util/DMaaPAuthFilter.java
index 574e9e1..8453a12 100644
--- a/src/main/java/com/att/nsa/dmaap/util/DMaaPAuthFilter.java
+++ b/src/main/java/com/att/nsa/dmaap/util/DMaaPAuthFilter.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -29,135 +29,60 @@ import javax.servlet.ServletRequest;
 import javax.servlet.ServletResponse;
 import javax.servlet.http.HttpServletRequest;
 
-//import com.att.eelf.configuration.EELFLogger;
-//import com.att.eelf.configuration.EELFManager;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
 import org.springframework.stereotype.Component;
 
 import com.att.cadi.filter.CadiFilter;
+//import ajsc.external.plugins.cadi.AjscCadiFilter;
 import javax.servlet.FilterConfig;
 
 /**
- * This is a Servlet Filter class overriding the AjscCadiFilter
- */
-@Component
-public class DMaaPAuthFilter extends CadiFilter {
-
-	// private Logger log = Logger.getLogger(DMaaPAuthFilter.class.toString());
-
-	// private static final EELFLogger log =
-	// EELFManager.getInstance().getLogger(DMaaPAuthFilter.class);
-	private Logger log = LoggerFactory.getLogger(DMaaPAuthFilter.class);
-
-	final Boolean enabled = "authentication-scheme-1".equalsIgnoreCase(System.getProperty("CadiAuthN"));
-
-	/**
-	 * This method will disable Cadi Authentication if cambria headers are
-	 * present in the request else continue with Cadi Authentication
+	 * This is a Servlet Filter class
+	 * overriding the AjscCadiFilter
 	 */
-	public void init(FilterConfig filterConfig) throws ServletException {
-
-		try {
-
-			super.init(filterConfig);
-
-		} catch (Exception ex) {
-			log.error("Ajsc Cadi Filter Exception" + ex);
-
-		}
-	}
-
-	@Override
-	public void doFilter(ServletRequest req, ServletResponse res, FilterChain chain)
-			throws IOException, ServletException {
-
-		log.info("inside servlet filter Cambria Auth Headers checking before doing other Authentication");
-		HttpServletRequest request = (HttpServletRequest) req;
-
-		boolean forceAAF = Boolean.valueOf(System.getProperty("forceAAF"));
-		if (forceAAF || 
-			null != request.getHeader("Authorization") || 
-			(null != request.getHeader("AppName") && 
-				request.getHeader("AppName").equalsIgnoreCase("invenio") )) {
+@Component	
+public class DMaaPAuthFilter extends CadiFilter {
+	
+		//private Logger log = Logger.getLogger(DMaaPAuthFilter.class.toString());
 
-			if (!enabled || 
-					request.getMethod().equalsIgnoreCase("head") || 
-					request.getHeader("DME2HealthCheck") != null) {
+		private static final EELFLogger log = EELFManager.getInstance().getLogger(DMaaPAuthFilter.class);
 				
-				chain.doFilter(req, res);
-				
-			} else {
-				
-				super.doFilter(req, res, chain);
-				
-			}
-		} else {
-
-			System.setProperty("CadiAuthN", "authentication-scheme-2");
-			chain.doFilter(req, res);
-
+		public DMaaPAuthFilter() throws Exception {
+			super();
 		}
-
-	}
-
-	@Override
-	public void log(Exception e, Object... elements) {
-		// TODO Auto-generated method stub
-		// super.log(e, elements);
-		// System.out.println(convertArrayToString(elements));
-		log.error(convertArrayToString(elements), e);
-
-	}
-
-	@Override
-	public void log(Level level, Object... elements) {
-
-		// System.out.println(willWrite().compareTo(level) );
-		if (willWrite().compareTo(level) <= 0) {
-			switch (level) {
-			case DEBUG:
-				log.debug(convertArrayToString(elements));
-				break;
-			case INFO:
-				log.info(convertArrayToString(elements));
-				break;
-			case ERROR:
-				log.error(convertArrayToString(elements));
-				break;
-			case AUDIT:
-				log.info(convertArrayToString(elements));
-				break;
-			case INIT:
-				log.info(convertArrayToString(elements));
-				break;
-			case WARN:
-				log.warn(convertArrayToString(elements));
-				break;
-			default:
-
-				log.warn(convertArrayToString(elements));
-
-			}
+		
+	/*	public void init(FilterConfig filterConfig) throws ServletException {
+              
+			super.init(filterConfig);
+              System.out.println("---------------------------- in init method");
+		}*/
+		
+		/**
+		 * This method will disable Cadi Authentication 
+		 * if cambria headers are present in the request
+		 * else continue with Cadi Authentication
+		 */
+		@Override
+		public void doFilter(ServletRequest req, ServletResponse res, FilterChain chain) throws IOException,
+				ServletException {
+		log.info("inside servlet filter Cambria Auth Headers checking before doing other Authentication");
+			HttpServletRequest request = (HttpServletRequest) req;
+				boolean forceAAF = Boolean.valueOf(System.getProperty("forceAAF"));
+				//if (forceAAF || null != request.getHeader("Authorization") ){
+					if (forceAAF || null != request.getHeader("Authorization") || 
+							(null != request.getHeader("AppName") &&  request.getHeader("AppName").equalsIgnoreCase("invenio") &&
+							 null != request.getHeader("cookie"))){
+						super.doFilter(req, res, chain);
+						
+				} else { 
+					System.setProperty("CadiAuthN", "authentication-scheme-2");
+						chain.doFilter(req, res);
+					
+					
+				} 
 
 		}
 
 	}
 
-	private String convertArrayToString(Object[] elements) {
-
-		StringBuilder strBuilder = new StringBuilder();
-		for (int i = 0; i < elements.length; i++) {
-			if (elements[i] instanceof String)
-				strBuilder.append((String) elements[i]);
-			else if (elements[i] instanceof Integer)
-				strBuilder.append((Integer) elements[i]);
-			else
-				strBuilder.append(elements[i]);
-		}
-		String newString = strBuilder.toString();
-		return newString;
-	}
-
-}
diff --git a/src/main/java/com/att/nsa/dmaap/util/ServicePropertiesMapBean.java b/src/main/java/com/att/nsa/dmaap/util/ServicePropertiesMapBean.java
index c5173c1..b16162b 100644
--- a/src/main/java/com/att/nsa/dmaap/util/ServicePropertiesMapBean.java
+++ b/src/main/java/com/att/nsa/dmaap/util/ServicePropertiesMapBean.java
@@ -8,14 +8,14 @@
  *  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.
  *  ============LICENSE_END=========================================================
- *
+ *  
  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  *  
  *******************************************************************************/
@@ -25,7 +25,7 @@ import com.att.nsa.dmaap.filemonitor.ServicePropertiesMap;
 
 /**
  * Class ServicePropertiesMapBean
- * @author author
+ * @author rajashree.khare
  *
  */
 public class ServicePropertiesMapBean {