update the package name1.1.11

Issue-ID: DMAAP-858
Change-Id: I49ae6eb9c51a261b64b911e607fcbbca46c5423c
Signed-off-by: sunil unnava <sunil.unnava@att.com>

11 files changed, 0 insertions, 2172 deletions

diff --git a/src/main/java/com/att/dmf/mr/beans/ApiKeyBean.java b/src/main/java/com/att/dmf/mr/beans/ApiKeyBean.java
deleted file mode 100644
index 4f0108f..0000000
--- a/src/main/java/com/att/dmf/mr/beans/ApiKeyBean.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*******************************************************************************
- *  ============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.dmf.mr.beans;
-
-import java.io.Serializable;
-
-import javax.xml.bind.annotation.XmlRootElement;
-
-import com.att.nsa.drumlin.till.data.uniqueStringGenerator;
-/**
- * 
- * @author anowarul.islam
- *
- */
-@XmlRootElement
-public class ApiKeyBean implements Serializable {
-
-	private static final long serialVersionUID = -8219849086890567740L;
-	
-	private static final String KEY_CHARS = "ABCDEFGHJIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789";
-	
-	private String email;
-	private String description;
-	/**
-	 * constructor
-	 */
-	public ApiKeyBean() {
-		super();
-	}
-/**
- * 
- * @param email
- * @param description
- */
-	public ApiKeyBean(String email, String description) {
-		super();
-		this.email = email;
-		this.description = description;
-	}
-
-	public String getEmail() {
-		return email;
-	}
-
-	public void setEmail(String email) {
-		this.email = email;
-	}
-
-	public String getDescription() {
-		return description;
-	}
-
-	public void setDescription(String description) {
-		this.description = description;
-	}
-
-	public String getKey() {
-		return generateKey(16);
-	}
-
-	public String getSharedSecret() {
-		return generateKey(24);
-	}
-	
-	private static String generateKey ( int length  ) {
-		return uniqueStringGenerator.createKeyUsingAlphabet ( KEY_CHARS, length );
-	}
-	
-}
diff --git a/src/main/java/com/att/dmf/mr/beans/DMaaPCambriaLimiter.java b/src/main/java/com/att/dmf/mr/beans/DMaaPCambriaLimiter.java
deleted file mode 100644
index 8cbf64f..0000000
--- a/src/main/java/com/att/dmf/mr/beans/DMaaPCambriaLimiter.java
+++ /dev/null
@@ -1,288 +0,0 @@
-/*******************************************************************************
- *  ============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.dmf.mr.beans;
-
-import java.util.Date;
-import java.util.HashMap;
-import java.util.concurrent.TimeUnit;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.stereotype.Component;
-
-import com.att.dmf.mr.CambriaApiException;
-import com.att.dmf.mr.constants.CambriaConstants;
-import com.att.dmf.mr.exception.DMaaPResponseCode;
-import com.att.dmf.mr.exception.ErrorResponse;
-import com.att.dmf.mr.utils.Utils;
-
-import com.att.eelf.configuration.EELFLogger;
-import com.att.eelf.configuration.EELFManager;
-import com.att.nsa.drumlin.service.standards.HttpStatusCodes;
-import com.att.nsa.drumlin.till.nv.rrNvReadable;
-import com.att.nsa.drumlin.till.nv.rrNvReadable.invalidSettingValue;
-import com.att.nsa.drumlin.till.nv.rrNvReadable.missingReqdSetting;
-import com.att.nsa.metrics.impl.CdmRateTicker;
-
-/**
- * class provide rate information
- * 
- * @author anowarul.islam
- *
- */
-@Component
-public class DMaaPCambriaLimiter {
-	private final HashMap<String, RateInfo> fRateInfo;
-	private final HashMap<String, RateInfoCheck> fRateInfoCheck;
-	private final double fMaxEmptyPollsPerMinute;
-	private final double fMaxPollsPerMinute;
-	private final int fWindowLengthMins;
-	private final long fSleepMs;
-	private final long fSleepMs1;
-	private static final EELFLogger log = EELFManager.getInstance().getLogger(DMaaPCambriaLimiter.class);
-	
-	/**
-	 * constructor initializes
-	 * 
-	 * @param settings
-	 * @throws missingReqdSetting
-	 * @throws invalidSettingValue
-	 */
-	@Autowired
-	public DMaaPCambriaLimiter(@Qualifier("propertyReader") rrNvReadable settings) {
-			fRateInfo = new HashMap<>();
-		fRateInfoCheck = new HashMap<>();
-		fMaxEmptyPollsPerMinute = settings.getDouble(CambriaConstants.kSetting_MaxEmptyPollsPerMinute,
-				CambriaConstants.kDefault_MaxEmptyPollsPerMinute);
-		fMaxPollsPerMinute = settings.getDouble(CambriaConstants.kSetting_MaxPollsPerMinute,
-				30);
-		fWindowLengthMins = settings.getInt(CambriaConstants.kSetting_RateLimitWindowLength,
-				CambriaConstants.kDefault_RateLimitWindowLength);
-		fSleepMs = settings.getLong(CambriaConstants.kSetting_SleepMsOnRateLimit,
-				CambriaConstants.kDefault_SleepMsOnRateLimit);
-		fSleepMs1 = settings.getLong(CambriaConstants.kSetting_SleepMsRealOnRateLimit,
-				5000);
-		
-	}
-	
-	/**
-	 * Construct a rate limiter.
-	 * 
-	 * @param maxEmptyPollsPerMinute
-	 *            Pass <= 0 to deactivate rate limiting.
-	 *            @param windowLengthMins
-	 */
-	public DMaaPCambriaLimiter(double maxEmptyPollsPerMinute, double maxPollsPerMinute,int windowLengthMins) {
-		this(maxEmptyPollsPerMinute,maxPollsPerMinute, windowLengthMins, getSleepMsForRate(maxEmptyPollsPerMinute),getSleepMsForRate(1));
-	}
-
-	/**
-	 * Construct a rate limiter
-	 * 
-	 * @param maxEmptyPollsPerMinute
-	 *            Pass <= 0 to deactivate rate limiting.
-	 * @param sleepMs
-	 * @param windowLengthMins
-	 */
-	public DMaaPCambriaLimiter(double maxEmptyPollsPerMinute,double maxPollsPerMinute, int windowLengthMins, long sleepMs ,long sleepMS1) {
-		fRateInfo = new HashMap<>();
-		fRateInfoCheck = new HashMap<>();
-		fMaxEmptyPollsPerMinute = Math.max(0, maxEmptyPollsPerMinute);
-		fMaxPollsPerMinute = Math.max(0, maxPollsPerMinute);
-		fWindowLengthMins = windowLengthMins;
-		fSleepMs = Math.max(0, sleepMs);
-		fSleepMs1 = Math.max(0, sleepMS1);
-	}
-
-	/**
-	 * static method provide the sleep time
-	 * 
-	 * @param ratePerMinute
-	 * @return
-	 */
-	public static long getSleepMsForRate(double ratePerMinute) {
-		if (ratePerMinute <= 0.0)
-			return 0;
-		return Math.max(1000, Math.round(60 * 1000 / ratePerMinute));
-	}
-
-	/**
-	 * Tell the rate limiter about a call to a topic/group/id. If the rate is
-	 * too high, this call delays its return and throws an exception.
-	 * 
-	 * @param topic
-	 * @param consumerGroup
-	 * @param clientId
-	 * @throws CambriaApiException
-	 */
-	public void onCall(String topic, String consumerGroup, String clientId,String remoteHost) throws CambriaApiException {
-		// do nothing if rate is configured 0 or less
-		if (fMaxEmptyPollsPerMinute <= 0) {
-			return;
-		}
-				// setup rate info for this tuple
-		final RateInfo ri = getRateInfo(topic, consumerGroup, clientId);
-		final double rate = ri.onCall();
-		log.info(ri.getLabel() + ": " + rate + " empty replies/minute.");
-		if (rate > fMaxEmptyPollsPerMinute) {
-			try {
-				log.warn(ri.getLabel() + ": " + rate + " empty replies/minute, limit is " + fMaxPollsPerMinute
-						+ ".");
-				if (fSleepMs > 0) {
-					log.warn(ri.getLabel() + ": " + "Slowing response with " + fSleepMs
-							+ " ms sleep, then responding in error.");
-					Thread.sleep(fSleepMs);
-					
-				} else {
-					log.info(ri.getLabel() + ": " + "No sleep configured, just throwing error.");
-				}
-			} catch (InterruptedException e) {
-				log.error("Exception "+ e);
-				// ignore
-			}
-			
-			
-			ErrorResponse errRes = new ErrorResponse(HttpStatusCodes.k429_tooManyRequests, 
-					DMaaPResponseCode.TOO_MANY_REQUESTS.getResponseCode(), 
-					"This client is making too many requests. Please use a long poll "
-							+ "setting to decrease the number of requests that result in empty responses. ","",Utils.getFormattedDate(new Date()),topic,"","",consumerGroup+"/"+clientId,remoteHost);
-			
-			log.info(errRes.toString());
-			throw new CambriaApiException(errRes);
-		}
-		
-		
-	}
-
-	/**
-	 * 
-	 * @param topic
-	 * @param consumerGroup
-	 * @param clientId
-	 * @param sentCount
-	 */
-	public void onSend(String topic, String consumerGroup, String clientId, long sentCount) {
-		// check for good replies
-		if (sentCount > 0) {
-			// that was a good send, reset the metric
-			getRateInfo(topic, consumerGroup, clientId).reset();
-		}
-	}
-
-	private static class RateInfo {
-		private final String fLabel;
-		private final CdmRateTicker fCallRateSinceLastMsgSend;
-		/**
-		 * constructor initialzes
-		 * 
-		 * @param label
-		 * @param windowLengthMinutes
-		 */
-		public RateInfo(String label, int windowLengthMinutes) {
-			fLabel = label;
-			fCallRateSinceLastMsgSend = new CdmRateTicker("Call rate since last msg send", 1, TimeUnit.MINUTES,
-					windowLengthMinutes, TimeUnit.MINUTES);
-		}
-		
-		public String getLabel() {
-			return fLabel;
-		}
-
-		/**
-		 * CdmRateTicker is reset
-		 */
-		public void reset() {
-			fCallRateSinceLastMsgSend.reset();
-		}
-
-		/**
-		 * 
-		 * @return
-		 */
-		public double onCall() {
-			fCallRateSinceLastMsgSend.tick();
-			return fCallRateSinceLastMsgSend.getRate();
-		}
-	}
-	
-	
-	
-	private static class RateInfoCheck {
-		
-		private final String fLabel;
-		private final CdmRateTicker fCallRateSinceLastMsgSend;
-		/**
-		 * constructor initialzes
-		 * 
-		 * @param label
-		 * @param windowLengthMinutes
-		 */
-		public RateInfoCheck(String label, int windowLengthMinutes) {
-			fLabel = label;
-			fCallRateSinceLastMsgSend = new CdmRateTicker("Call rate since last msg send", 1, TimeUnit.MINUTES,
-					windowLengthMinutes, TimeUnit.MINUTES);
-		}
-
-		public String getLabel() {
-			return fLabel;
-		}
-
-		/**
-		 * CdmRateTicker is reset
-		 */
-		public void reset() {
-			fCallRateSinceLastMsgSend.reset();
-		}
-
-		/**
-		 * 
-		 * @return
-		 */
-		public double onCall() {
-			fCallRateSinceLastMsgSend.tick();
-			return fCallRateSinceLastMsgSend.getRate();
-		}
-	}
-	
-	
-	
-	
-	private RateInfo getRateInfo(String topic, String consumerGroup, String clientId) {
-		final String key = makeKey(topic, consumerGroup, clientId);
-		RateInfo ri = fRateInfo.get(key);
-		if (ri == null) {
-			ri = new RateInfo(key, fWindowLengthMins);
-			fRateInfo.put(key, ri);
-		}
-		return ri;
-	}
-	
-	
-	
-
-	
-	
-	
-	private String makeKey(String topic, String group, String id) {
-		return topic + "::" + group + "::" + id;
-	}
-}
diff --git a/src/main/java/com/att/dmf/mr/beans/DMaaPContext.java b/src/main/java/com/att/dmf/mr/beans/DMaaPContext.java
deleted file mode 100644
index a880877..0000000
--- a/src/main/java/com/att/dmf/mr/beans/DMaaPContext.java
+++ /dev/null
@@ -1,104 +0,0 @@
-/*******************************************************************************
- *  ============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.dmf.mr.beans;
-
-import java.text.SimpleDateFormat;
-import java.util.Date;
-
-import javax.servlet.http.HttpServletRequest;
-import javax.servlet.http.HttpServletResponse;
-import javax.servlet.http.HttpSession;
-
-import com.att.dmf.mr.utils.ConfigurationReader;
-
-/**
- * DMaaPContext provide and maintain all the configurations , Http request/response
- * Session and consumer Request Time
- * @author nilanjana.maity
- *
- */
-public class DMaaPContext {
-
-    private ConfigurationReader configReader;
-    private HttpServletRequest request;
-    private HttpServletResponse response;
-    private HttpSession session;
-    private String consumerRequestTime;
-    static int i=0;
-    
-    public synchronized static long getBatchID() {
-    	try{
-    		final long metricsSendTime = System.currentTimeMillis();
-    		final Date d = new Date(metricsSendTime);
-    		final String text = new SimpleDateFormat("ddMMyyyyHHmmss").format(d);
-    		long dt= Long.valueOf(text)+i;
-    		i++;
-    		return dt;
-    	}
-    	catch(NumberFormatException ex){
-    		return 0;
-    	}
-    }
-    
-    public HttpServletRequest getRequest() {
-        return request;
-    }
-
-    public void setRequest(HttpServletRequest request) {
-        this.request = request;
-    }
-
-    public HttpServletResponse getResponse() {
-        return response;
-    }
-
-    public void setResponse(HttpServletResponse response) {
-        this.response = response;
-    }
-
-    public HttpSession getSession() {
-        this.session = request.getSession();
-        return session;
-    }
-
-    public void setSession(HttpSession session) {
-        this.session = session;
-    }
-
-    public ConfigurationReader getConfigReader() {
-    	  return configReader;
-    }
-
-    public void setConfigReader(ConfigurationReader configReader) {
-        this.configReader = configReader;
-    }
-
-    public String getConsumerRequestTime() {
-        return consumerRequestTime;
-    }
-
-    public void setConsumerRequestTime(String consumerRequestTime) {
-        this.consumerRequestTime = consumerRequestTime;
-    }
-    
-    
-}
diff --git a/src/main/java/com/att/dmf/mr/beans/DMaaPKafkaConsumerFactory.java b/src/main/java/com/att/dmf/mr/beans/DMaaPKafkaConsumerFactory.java
deleted file mode 100644
index fb0ace0..0000000
--- a/src/main/java/com/att/dmf/mr/beans/DMaaPKafkaConsumerFactory.java
+++ /dev/null
@@ -1,361 +0,0 @@
-/*******************************************************************************
- *  ============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.dmf.mr.beans;
-
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Properties;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.recipes.locks.InterProcessMutex;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.springframework.beans.factory.annotation.Qualifier;
-
-import com.att.ajsc.filemonitor.AJSCPropertiesMap;
-import com.att.dmf.mr.CambriaApiException;
-import com.att.dmf.mr.backends.Consumer;
-import com.att.dmf.mr.backends.ConsumerFactory;
-import com.att.dmf.mr.backends.MetricsSet;
-import com.att.dmf.mr.backends.kafka.Kafka011Consumer;
-import com.att.dmf.mr.backends.kafka.Kafka011ConsumerUtil;
-import com.att.dmf.mr.backends.kafka.KafkaConsumerCache;
-import com.att.dmf.mr.backends.kafka.KafkaConsumerCache.KafkaConsumerCacheException;
-import com.att.dmf.mr.backends.kafka.KafkaLiveLockAvoider2;
-import com.att.dmf.mr.backends.kafka.LiveLockAvoidance;
-import com.att.dmf.mr.constants.CambriaConstants;
-import com.att.dmf.mr.utils.ConfigurationReader;
-import com.att.dmf.mr.utils.Utils;
-import com.att.eelf.configuration.EELFLogger;
-import com.att.eelf.configuration.EELFManager;
-import com.att.nsa.drumlin.till.nv.rrNvReadable.missingReqdSetting;
-
-/**
- * @author nilanjana.maity
- *
- */
-public class DMaaPKafkaConsumerFactory implements ConsumerFactory {
-
-	
-	private static final EELFLogger log = EELFManager.getInstance().getLogger(DMaaPKafkaConsumerFactory.class);
-	
-
-	/**
-	 * constructor initialization
-	 * 
-	 * @param settings
-	 * @param metrics
-	 * @param curator
-	 * @throws missingReqdSetting
-	 * @throws KafkaConsumerCacheException
-	 * @throws UnknownHostException
-	 */
-
-	public DMaaPKafkaConsumerFactory(@Qualifier("dMaaPMetricsSet") MetricsSet metrics,
-			@Qualifier("curator") CuratorFramework curator,
-			@Qualifier("kafkalockavoid") KafkaLiveLockAvoider2 kafkaLiveLockAvoider)
-			throws missingReqdSetting, KafkaConsumerCacheException, UnknownHostException {
-
-		String apiNodeId = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
-				CambriaConstants.kSetting_ApiNodeIdentifier);
-		if (apiNodeId == null) {
-
-			apiNodeId = InetAddress.getLocalHost().getCanonicalHostName() + ":" + CambriaConstants.kDefault_Port;
-		}
-
-		log.info("This Cambria API Node identifies itself as [" + apiNodeId + "].");
-		final String mode = CambriaConstants.DMAAP;
-
-		fkafkaBrokers = com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
-				"kafka.metadata.broker.list");
-		if (null == fkafkaBrokers) {
-
-			fkafkaBrokers = "localhost:9092";
-		}
-
-		boolean kSetting_EnableCache = kDefault_IsCacheEnabled;
-		String strkSetting_EnableCache = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
-				"cambria.consumer.cache.enabled");
-		if (null != strkSetting_EnableCache)
-			kSetting_EnableCache = Boolean.parseBoolean(strkSetting_EnableCache);
-
-		final boolean isCacheEnabled = kSetting_EnableCache;
-
-		
-		fCache = null;
-		if (isCacheEnabled) {
-			fCache = KafkaConsumerCache.getInstance();
-
-		}
-		if (fCache != null) {
-			fCache.setfMetrics(metrics);
-			fCache.setfApiId(apiNodeId);
-			fCache.startCache(mode, curator);
-			if(kafkaLiveLockAvoider!=null){
-			kafkaLiveLockAvoider.startNewWatcherForServer(apiNodeId, makeAvoidanceCallback(apiNodeId));
-			fkafkaLiveLockAvoider = kafkaLiveLockAvoider;
-			}
-		}
-	}
-
-	/*
-	 * getConsumerFor
-	 * 
-	 * @see
-	 * com.att.dmf.mr.backends.ConsumerFactory#getConsumerFor(java.lang.String,
-	 * java.lang.String, java.lang.String, int, java.lang.String) This method is
-	 * used by EventServiceImpl.getEvents() method to get a Kakfa consumer
-	 * either from kafkaconsumer cache or create a new connection This also get
-	 * the list of other consumer objects for the same consumer group and set to
-	 * KafkaConsumer object. This list may be used during poll-rebalancing
-	 * issue.
-	 */
-	@Override
-	public Consumer getConsumerFor(String topic, String consumerGroupName, String consumerId, int timeoutMs,
-			String remotehost) throws UnavailableException, CambriaApiException {
-		Kafka011Consumer kc;
-
-		// To synchronize based on the consumer group.
-
-		Object syncObject = synchash.get(topic + consumerGroupName);
-		if (null == syncObject) {
-			syncObject = new Object();
-			synchash.put(topic + consumerGroupName, syncObject);
-		}
-
-		synchronized (syncObject) {
-			try {
-				kc = (fCache != null) ? fCache.getConsumerFor(topic, consumerGroupName, consumerId) : null; // consumerId
-
-			} catch (KafkaConsumerCacheException e) {
-				log.info("######@@@@### Error occured in Kafka Caching" + e + "  " + topic + "::" + consumerGroupName
-						+ "::" + consumerId);
-				log.error("####@@@@## Error occured in Kafka Caching" + e + "  " + topic + "::" + consumerGroupName
-						+ "::" + consumerId);
-				throw new UnavailableException(e);
-			}
-
-			// Ideally if cache exists below flow should be skipped. If cache
-			// didnt
-			// exist, then create this first time on this node.
-			if (kc == null) {
-
-				log.info("^Kafka consumer cache value " + topic + "::" + consumerGroupName + "::" + consumerId + " =>"
-						+ kc);
-
-				final InterProcessMutex ipLock = new InterProcessMutex(ConfigurationReader.getCurator(),
-						"/consumerFactory/" + topic + "/" + consumerGroupName + "/" + consumerId);
-				boolean locked = false;
-
-				try {
-
-					locked = ipLock.acquire(30, TimeUnit.SECONDS);
-					if (!locked) {
-
-						log.info("Could not acquire lock in order to create (topic, group, consumer) = " + "(" + topic
-								+ ", " + consumerGroupName + ", " + consumerId + ") from " + remotehost);
-						throw new UnavailableException(
-								"Could not acquire lock in order to create (topic, group, consumer) = " + "(" + topic
-										+ ", " + consumerGroupName + ", " + consumerId + ") " + remotehost);
-					}
-
-					// ConfigurationReader.getCurator().checkExists().forPath("S").
-
-					log.info("Creating Kafka consumer for group [" + consumerGroupName + "], consumer [" + consumerId
-							+ "], on topic [" + topic + "].");
-					
-					if (fCache != null) {
-						fCache.signalOwnership(topic, consumerGroupName, consumerId);
-					}
-					
-					final Properties props = createConsumerConfig(topic,consumerGroupName, consumerId);
-					long fCreateTimeMs = System.currentTimeMillis();
-					KafkaConsumer<String, String> cc = new KafkaConsumer<>(props);
-					kc = new Kafka011Consumer(topic, consumerGroupName, consumerId, cc, fkafkaLiveLockAvoider);
-					log.info(" kafka stream created in " + (System.currentTimeMillis() - fCreateTimeMs));
-
-					if (fCache != null) {
-						fCache.putConsumerFor(topic, consumerGroupName, consumerId, kc); //
-					}
-
-				} catch (org.I0Itec.zkclient.exception.ZkTimeoutException x) {
-					log.info(
-							"Kafka consumer couldn't connect to ZK. " + x + " " + consumerGroupName + "/" + consumerId);
-					throw new UnavailableException("Couldn't connect to ZK.");
-				} catch (KafkaConsumerCacheException e) {
-					log.info("Failed to cache consumer (this may have performance implications): " + e.getMessage()
-							+ " " + consumerGroupName + "/" + consumerId);
-				} catch (UnavailableException u) {
-					log.info("Failed and in UnavailableException block " + u.getMessage() + " " + consumerGroupName
-							+ "/" + consumerId);
-					throw new UnavailableException("Error while acquiring consumer factory lock " + u.getMessage(), u);
-				} catch (Exception e) {
-					log.info("Failed and go to Exception block " + e.getMessage() + " " + consumerGroupName + "/"
-							+ consumerId);
-					log.error("Failed and go to Exception block " + e.getMessage() + " " + consumerGroupName + "/"
-							+ consumerId);
-					
-				} finally {
-					if (locked) {
-						try {
-							ipLock.release();
-						} catch (Exception e) {
-							throw new UnavailableException("Error while releasing consumer factory lock" + e, e);
-						}
-					}
-				}
-			}
-		}
-		return kc;
-	}
-
-	@Override
-	public synchronized void destroyConsumer(String topic, String consumerGroup, String clientId) {
-		if (fCache != null) {
-			fCache.dropConsumer(topic, consumerGroup, clientId);
-		}
-	}
-
-	@Override
-	public synchronized Collection<? extends Consumer> getConsumers() {
-		return fCache.getConsumers();
-	}
-
-	@Override
-	public synchronized void dropCache() {
-		fCache.dropAllConsumers();
-	}
-
-	
-	private KafkaConsumerCache fCache;
-	private KafkaLiveLockAvoider2 fkafkaLiveLockAvoider;
-	private String fkafkaBrokers;
-
-
-
-	private static String makeLongKey(String key, String prefix) {
-		return prefix + "." + key;
-	}
-
-	private void transferSettingIfProvided(Properties target, String key, String prefix) {
-		String keyVal = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, makeLongKey(key, prefix));
-
-		
-		if (null != keyVal) {
-		
-			log.info("Setting [" + key + "] to " + keyVal + ".");
-			target.put(key, keyVal);
-		}
-	}
-
-	/**
-	 * Name CreateConsumerconfig  
-	 * @param topic
-	 * @param groupId
-	 * @param consumerId
-	 * @return Properties
-	 * 
-	 * This method is to create Properties required to create kafka connection
-	 * Group name is replaced with different format groupid--topic to address same 
-	 * groupids for multiple topics. Same groupid with multiple topics 
-	 * may start frequent consumer rebalancing on all the topics . Replacing them makes it unique
-	 */
-	private Properties createConsumerConfig(String topic ,String groupId, String consumerId) {
-		final Properties props = new Properties();
-		//fakeGroupName is added to avoid multiple consumer group for multiple topics.Donot Change this logic
-		//Fix for CPFMF-644 : 
-		final String fakeGroupName = groupId + "--" + topic;
-		props.put("group.id", fakeGroupName);
-		props.put("enable.auto.commit", "false"); // 0.11
-		props.put("bootstrap.servers", fkafkaBrokers);
-		if(Utils.isCadiEnabled()){
-		props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='"+Utils.getKafkaproperty()+"';");
-		props.put("security.protocol", "SASL_PLAINTEXT");
-		props.put("sasl.mechanism", "PLAIN");
-		}
-		props.put("client.id", consumerId);
-
-		// additional settings: start with our defaults, then pull in configured
-		// overrides
-		populateKafkaInternalDefaultsMap();
-		for (String key : KafkaConsumerKeys) {
-			transferSettingIfProvided(props, key, "kafka");
-		}
-
-		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
-		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
-
-		return props;
-	}
-
-
-	private static final String KafkaConsumerKeys[] = { "bootstrap.servers", "heartbeat.interval.ms",
-			"auto.offset.reset", "exclude.internal.topics", "session.timeout.ms", "fetch.max.bytes",
-			"auto.commit.interval.ms", "connections.max.idle.ms", "fetch.min.bytes", "isolation.level",
-			"fetch.max.bytes", "request.timeout.ms", "fetch.max.wait.bytes", "reconnect.backoff.max.ms",
-			"max.partition.fetch.bytes", "reconnect.backoff.max.ms", "reconnect.backoff.ms", "retry.backoff.ms",
-			"max.poll.interval.ms", "max.poll.records", "receive.buffer.bytes", "metadata.max.age.ms" };
-
-	/**
-	 * putting values in hashmap like consumer timeout, zookeeper time out, etc
-	 * 
-	 * @param setting
-	 */
-	private static void populateKafkaInternalDefaultsMap() { }
-
-	/*
-	 * The starterIncremnt value is just to emulate calling certain consumers,
-	 * in this test app all the consumers are local
-	 * 
-	 */
-	private LiveLockAvoidance makeAvoidanceCallback(final String appId) {
-
-		return new LiveLockAvoidance() {
-
-			@Override
-			public String getAppId() {
-				return appId;
-			}
-
-			@Override
-			public void handleRebalanceUnlock(String groupName) {
-				log.info("FORCE A POLL NOW FOR appId: [{}] group: [{}]", getAppId(), groupName);
-				Kafka011ConsumerUtil.forcePollOnConsumer(groupName + "::");
-			}
-
-		};
-
-	}
-
-	@SuppressWarnings("rawtypes")
-	@Override
-	public HashMap getConsumerForKafka011(String topic, String consumerGroupName, String consumerId, int timeoutMs,
-			String remotehost) throws UnavailableException, CambriaApiException {
-		// TODO Auto-generated method stub
-		return null;
-	}
-
-	private HashMap<String, Object> synchash = new HashMap<String, Object>();
-
-}
\ No newline at end of file
diff --git a/src/main/java/com/att/dmf/mr/beans/DMaaPKafkaMetaBroker.java b/src/main/java/com/att/dmf/mr/beans/DMaaPKafkaMetaBroker.java
deleted file mode 100644
index acf4824..0000000
--- a/src/main/java/com/att/dmf/mr/beans/DMaaPKafkaMetaBroker.java
+++ /dev/null
@@ -1,495 +0,0 @@
-/*******************************************************************************
- *  ============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.dmf.mr.beans;
-
-import java.util.Arrays;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Properties;
-import java.util.Set;
-import java.util.TreeSet;
-import java.util.concurrent.ExecutionException;
-
-import org.I0Itec.zkclient.ZkClient;
-import org.I0Itec.zkclient.exception.ZkNoNodeException;
-import org.apache.kafka.clients.admin.AdminClient;
-import org.apache.kafka.clients.admin.AdminClientConfig;
-import org.apache.kafka.clients.admin.CreateTopicsResult;
-import org.apache.kafka.clients.admin.NewTopic;
-import org.apache.kafka.common.KafkaFuture;
-import org.json.JSONObject;
-import org.json.JSONArray;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.stereotype.Component;
-
-import com.att.dmf.mr.CambriaApiException;
-import com.att.dmf.mr.constants.CambriaConstants;
-import com.att.dmf.mr.metabroker.Broker;
-import com.att.dmf.mr.metabroker.Broker1;
-import com.att.dmf.mr.metabroker.Topic;
-import com.att.dmf.mr.utils.ConfigurationReader;
-import com.att.dmf.mr.utils.Utils;
-//import org.apache.log4-j.Logger;
-import com.att.eelf.configuration.EELFLogger;
-import com.att.eelf.configuration.EELFManager;
-
-import com.att.nsa.configs.ConfigDb;
-import com.att.nsa.configs.ConfigDbException;
-import com.att.nsa.configs.ConfigPath;
-import com.att.nsa.drumlin.service.standards.HttpStatusCodes;
-import com.att.nsa.drumlin.till.nv.rrNvReadable;
-import com.att.nsa.security.NsaAcl;
-import com.att.nsa.security.NsaAclUtils;
-import com.att.nsa.security.NsaApiKey;
-
-
-/**
- * class performing all topic operations
- * 
- * @author anowarul.islam
- *
- */
-//@Component
-public class DMaaPKafkaMetaBroker implements Broker1 {
-
-	public DMaaPKafkaMetaBroker() {
-		fZk = null;
-		fCambriaConfig = null;
-		fBaseTopicData = null;
-		final Properties props = new Properties ();
-		String fkafkaBrokers = com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
-				"kafka.metadata.broker.list");
-		if (null == fkafkaBrokers) {
-
-			fkafkaBrokers = "localhost:9092";
-		}
-		
-	     props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, fkafkaBrokers );
-	     if(Utils.isCadiEnabled()){
-	     props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='"+Utils.getKafkaproperty()+"';");
-	  	 props.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");	     
-	     props.put("sasl.mechanism", "PLAIN");
-	     }
-	   
-	     fKafkaAdminClient=AdminClient.create ( props );
-	    
-	}
-
-	//private static final Logger log = Logger.getLogger(DMaaPKafkaMetaBroker.class);
-	private static final EELFLogger log = EELFManager.getInstance().getLogger(ConfigurationReader.class);
-	private final AdminClient fKafkaAdminClient;
-	
-	
-
-	/**
-	 * DMaaPKafkaMetaBroker constructor initializing
-	 * 
-	 * @param settings
-	 * @param zk
-	 * @param configDb
-	 */
-	public DMaaPKafkaMetaBroker(@Qualifier("propertyReader") rrNvReadable settings,
-			@Qualifier("dMaaPZkClient") ZkClient zk, @Qualifier("dMaaPZkConfigDb") ConfigDb configDb) {
-		//fSettings = settings;
-		fZk = zk;
-		fCambriaConfig = configDb;
-		fBaseTopicData = configDb.parse("/topics");
-		final Properties props = new Properties ();
-		String fkafkaBrokers = com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
-				"kafka.metadata.broker.list");
-		if (null == fkafkaBrokers) {
-
-			fkafkaBrokers = "localhost:9092";
-		}
-		
-		 if(Utils.isCadiEnabled()){
-		 props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='"+Utils.getKafkaproperty()+"';");
-	  	 props.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");	     
-	     props.put("sasl.mechanism", "PLAIN");
-		 }
-	     props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, fkafkaBrokers );
-	     
-	     fKafkaAdminClient=AdminClient.create ( props );
-	    
-		
-		
-	}
-	
-	public DMaaPKafkaMetaBroker( rrNvReadable settings,
-			ZkClient zk,  ConfigDb configDb,AdminClient client) {
-		
-		fZk = zk;
-		fCambriaConfig = configDb;
-		fBaseTopicData = configDb.parse("/topics");
-	    fKafkaAdminClient= client;
-	   
-		
-		
-	}
-
-	@Override
-	public List<Topic> getAllTopics() throws ConfigDbException {
-		log.info("Retrieving list of all the topics.");
-		final LinkedList<Topic> result = new LinkedList<Topic>();
-		try {
-			log.info("Retrieving all topics from root: " + zkTopicsRoot);
-			final List<String> topics = fZk.getChildren(zkTopicsRoot);
-			for (String topic : topics) {
-				result.add(new KafkaTopic(topic, fCambriaConfig, fBaseTopicData));
-			}
-			JSONObject dataObj = new JSONObject();
-			dataObj.put("topics", new JSONObject());
-
-			for (String topic : topics) {
-				dataObj.getJSONObject("topics").put(topic, new JSONObject());
-			}
-		} catch (ZkNoNodeException excp) {
-			// very fresh kafka doesn't have any topics or a topics node
-			log.error("ZK doesn't have a Kakfa topics node at " + zkTopicsRoot, excp);
-		}
-		return result;
-	}
-
-	@Override
-	public Topic getTopic(String topic) throws ConfigDbException {
-		if (fZk.exists(zkTopicsRoot + "/" + topic)) {
-			return getKafkaTopicConfig(fCambriaConfig, fBaseTopicData, topic);
-		}
-		// else: no such topic in kafka
-		return null;
-	}
-
-	/**
-	 * static method get KafkaTopic object
-	 * 
-	 * @param db
-	 * @param base
-	 * @param topic
-	 * @return
-	 * @throws ConfigDbException
-	 */
-	public static KafkaTopic getKafkaTopicConfig(ConfigDb db, ConfigPath base, String topic) throws ConfigDbException {
-		return new KafkaTopic(topic, db, base);
-	}
-
-	/**
-	 * creating topic
-	 */
-	@Override
-	public Topic createTopic(String topic, String desc, String ownerApiKey, int partitions, int replicas,
-			boolean transactionEnabled) throws TopicExistsException, CambriaApiException,ConfigDbException {
-		log.info("Creating topic: " + topic);
-		try {
-			log.info("Check if topic [" + topic + "] exist.");
-			// first check for existence "our way"
-			final Topic t = getTopic(topic);
-			if (t != null) {
-				log.info("Could not create topic [" + topic + "]. Topic Already exists.");
-				throw new TopicExistsException("Could not create topic [" + topic + "]. Topic Alreay exists.");
-			}
-		} catch (ConfigDbException e1) {
-			log.error("Topic [" + topic + "] could not be created. Couldn't check topic data in config db.", e1);
-			throw new CambriaApiException(HttpStatusCodes.k503_serviceUnavailable,
-					"Couldn't check topic data in config db.");
-		}
-
-		// we only allow 3 replicas. (If we don't test this, we get weird
-		// results from the cluster,
-		// so explicit test and fail.)
-		if (replicas < 1 || replicas > 3) {
-			log.info("Topic [" + topic + "] could not be created. The replica count must be between 1 and 3.");
-			throw new CambriaApiException(HttpStatusCodes.k400_badRequest,
-					"The replica count must be between 1 and 3.");
-		}
-		if (partitions < 1) {
-			log.info("Topic [" + topic + "] could not be created. The partition count must be at least 1.");
-			throw new CambriaApiException(HttpStatusCodes.k400_badRequest, "The partition count must be at least 1.");
-		}
-
-		// create via kafka
-		
-			try
-			{
-				final NewTopic topicRequest = new NewTopic ( topic, partitions, new Integer(replicas).shortValue () );
-				final CreateTopicsResult ctr = fKafkaAdminClient.createTopics ( Arrays.asList ( topicRequest ) );
-				final KafkaFuture<Void> ctrResult = ctr.all ();
-				ctrResult.get ();
-				// underlying Kafka topic created. now setup our API info
-				return createTopicEntry ( topic, desc, ownerApiKey, transactionEnabled );
-			}
-			catch ( InterruptedException e )
-			{
-				
-				log.warn ( "Execution of describeTopics timed out." );
-				throw new ConfigDbException ( e );
-			}
-			catch ( ExecutionException e )
-			{
-				
-				log.warn ( "Execution of describeTopics failed: " + e.getCause ().getMessage (), e.getCause () );
-				throw new ConfigDbException ( e.getCause () );
-			}
-		
-	}
-
-	@Override
-	public void deleteTopic(String topic) throws CambriaApiException, TopicExistsException,ConfigDbException {
-		log.info("Deleting topic: " + topic);
-		ZkClient zkClient = null;
-		try {
-			log.info("Loading zookeeper client for topic deletion.");
-					// topic creation. (Otherwise, the topic is only partially created
-			// in ZK.)
-			
-			
-			fKafkaAdminClient.deleteTopics(Arrays.asList(topic));
-			log.info("Zookeeper client loaded successfully. Deleting topic.");
-			
-		} catch (Exception e) {
-			log.error("Failed to delete topic [" + topic + "]. " + e.getMessage(), e);
-			throw new ConfigDbException(e);
-		}  finally {
-			log.info("Closing zookeeper connection.");
-			if (zkClient != null)
-				zkClient.close();
-		}
-
-		// throw new UnsupportedOperationException ( "We can't programmatically
-		// delete Kafka topics yet." );
-	}
-
-	//private final rrNvReadable fSettings;
-	private final ZkClient fZk;
-	private final ConfigDb fCambriaConfig;
-	private final ConfigPath fBaseTopicData;
-
-	private static final String zkTopicsRoot = "/brokers/topics";
-	private static final JSONObject kEmptyAcl = new JSONObject();
-
-	/**
-	 * method Providing KafkaTopic Object associated with owner and
-	 * transactionenabled or not
-	 * 
-	 * @param name
-	 * @param desc
-	 * @param owner
-	 * @param transactionEnabled
-	 * @return
-	 * @throws ConfigDbException
-	 */
-	public KafkaTopic createTopicEntry(String name, String desc, String owner, boolean transactionEnabled)
-			throws ConfigDbException {
-		return createTopicEntry(fCambriaConfig, fBaseTopicData, name, desc, owner, transactionEnabled);
-	}
-
-	/**
-	 * static method giving kafka topic object
-	 * 
-	 * @param db
-	 * @param basePath
-	 * @param name
-	 * @param desc
-	 * @param owner
-	 * @param transactionEnabled
-	 * @return
-	 * @throws ConfigDbException
-	 */
-	public static KafkaTopic createTopicEntry(ConfigDb db, ConfigPath basePath, String name, String desc, String owner,
-			boolean transactionEnabled) throws ConfigDbException {
-		final JSONObject o = new JSONObject();
-		o.put("owner", owner);
-		o.put("description", desc);
-		o.put("txenabled", transactionEnabled);
-		db.store(basePath.getChild(name), o.toString());
-		return new KafkaTopic(name, db, basePath);
-	}
-
-	/**
-	 * class performing all user opearation like user is eligible to read,
-	 * write. permitting a user to write and read,
-	 * 
-	 * @author anowarul.islam
-	 *
-	 */
-	public static class KafkaTopic implements Topic {
-		/**
-		 * constructor initializes
-		 * 
-		 * @param name
-		 * @param configdb
-		 * @param baseTopic
-		 * @throws ConfigDbException
-		 */
-		public KafkaTopic(String name, ConfigDb configdb, ConfigPath baseTopic) throws ConfigDbException {
-			fName = name;
-			fConfigDb = configdb;
-			fBaseTopicData = baseTopic;
-
-			String data = fConfigDb.load(fBaseTopicData.getChild(fName));
-			if (data == null) {
-				data = "{}";
-			}
-
-			final JSONObject o = new JSONObject(data);
-			fOwner = o.optString("owner", "");
-			fDesc = o.optString("description", "");
-			fTransactionEnabled = o.optBoolean("txenabled", false);// default
-																	// value is
-																	// false
-			// if this topic has an owner, it needs both read/write ACLs. If there's no
-						// owner (or it's empty), null is okay -- this is for existing or implicitly
-						// created topics.
-						JSONObject readers = o.optJSONObject ( "readers" );
-						if ( readers == null && fOwner.length () > 0 ) readers = kEmptyAcl;
-						fReaders =  fromJson ( readers );
-
-						JSONObject writers = o.optJSONObject ( "writers" );
-						if ( writers == null && fOwner.length () > 0 ) writers = kEmptyAcl;
-						fWriters = fromJson ( writers );
-		}
-		
-		private NsaAcl fromJson(JSONObject o) {
-			NsaAcl acl = new NsaAcl();
-			if (o != null) {
-				JSONArray a = o.optJSONArray("allowed");
-				if (a != null) {
-					for (int i = 0; i < a.length(); ++i) {
-						String user = a.getString(i);
-						acl.add(user);
-					}
-				}
-			}
-			return acl;
-		}
-
-		@Override
-		public String getName() {
-			return fName;
-		}
-
-		@Override
-		public String getOwner() {
-			return fOwner;
-		}
-
-		@Override
-		public String getDescription() {
-			return fDesc;
-		}
-
-		@Override
-		public NsaAcl getReaderAcl() {
-			return fReaders;
-		}
-
-		@Override
-		public NsaAcl getWriterAcl() {
-			return fWriters;
-		}
-
-		@Override
-		public void checkUserRead(NsaApiKey user) throws AccessDeniedException  {
-			NsaAclUtils.checkUserAccess ( fOwner, getReaderAcl(), user );
-		}
-
-		@Override
-		public void checkUserWrite(NsaApiKey user) throws AccessDeniedException  {
-			NsaAclUtils.checkUserAccess ( fOwner, getWriterAcl(), user );
-		}
-
-		@Override
-		public void permitWritesFromUser(String pubId, NsaApiKey asUser)
-				throws ConfigDbException, AccessDeniedException {
-			updateAcl(asUser, false, true, pubId);
-		}
-
-		@Override
-		public void denyWritesFromUser(String pubId, NsaApiKey asUser) throws ConfigDbException, AccessDeniedException {
-			updateAcl(asUser, false, false, pubId);
-		}
-
-		@Override
-		public void permitReadsByUser(String consumerId, NsaApiKey asUser)
-				throws ConfigDbException, AccessDeniedException {
-			updateAcl(asUser, true, true, consumerId);
-		}
-
-		@Override
-		public void denyReadsByUser(String consumerId, NsaApiKey asUser)
-				throws ConfigDbException, AccessDeniedException {
-			updateAcl(asUser, true, false, consumerId);
-		}
-
-		private void updateAcl(NsaApiKey asUser, boolean reader, boolean add, String key)
-				throws ConfigDbException, AccessDeniedException{
-			try
-			{
-				final NsaAcl acl = NsaAclUtils.updateAcl ( this, asUser, key, reader, add );
-	
-				// we have to assume we have current data, or load it again. for the expected use
-				// case, assuming we can overwrite the data is fine.
-				final JSONObject o = new JSONObject ();
-				o.put ( "owner", fOwner );
-				o.put ( "readers", safeSerialize ( reader ? acl : fReaders ) );
-				o.put ( "writers", safeSerialize ( reader ? fWriters : acl ) );
-				fConfigDb.store ( fBaseTopicData.getChild ( fName ), o.toString () );
-				
-				log.info ( "ACL_UPDATE: " + asUser.getKey () + " " + ( add ? "added" : "removed" ) + ( reader?"subscriber":"publisher" ) + " " + key + " on " + fName );
-	
-			}
-			catch ( ConfigDbException x )
-			{
-				throw x;
-			}
-			catch ( AccessDeniedException x )
-			{
-				throw x;
-			}
-			
-		}
-
-		private JSONObject safeSerialize(NsaAcl acl) {
-			return acl == null ? null : acl.serialize();
-		}
-
-		private final String fName;
-		private final ConfigDb fConfigDb;
-		private final ConfigPath fBaseTopicData;
-		private final String fOwner;
-		private final String fDesc;
-		private final NsaAcl fReaders;
-		private final NsaAcl fWriters;
-		private boolean fTransactionEnabled;
-	
-		public boolean isTransactionEnabled() {
-			return fTransactionEnabled;
-		}
-
-		@Override
-		public Set<String> getOwners() {
-			final TreeSet<String> owners = new TreeSet<String> ();
-			owners.add ( fOwner );
-			return owners;
-		}
-	}
-
-}
diff --git a/src/main/java/com/att/dmf/mr/beans/DMaaPMetricsSet.java b/src/main/java/com/att/dmf/mr/beans/DMaaPMetricsSet.java
deleted file mode 100644
index 4c9532b..0000000
--- a/src/main/java/com/att/dmf/mr/beans/DMaaPMetricsSet.java
+++ /dev/null
@@ -1,231 +0,0 @@
-/*******************************************************************************
- *  ============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.dmf.mr.beans;
-
-import java.text.SimpleDateFormat;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-
-import com.att.dmf.mr.CambriaApiVersionInfo;
-import com.att.dmf.mr.backends.MetricsSet;
-import com.att.mr.apiServer.metrics.cambria.DMaaPMetricsSender;
-import com.att.nsa.drumlin.till.nv.rrNvReadable;
-import com.att.nsa.metrics.impl.CdmConstant;
-import com.att.nsa.metrics.impl.CdmCounter;
-import com.att.nsa.metrics.impl.CdmMetricsRegistryImpl;
-import com.att.nsa.metrics.impl.CdmMovingAverage;
-import com.att.nsa.metrics.impl.CdmRateTicker;
-import com.att.nsa.metrics.impl.CdmSimpleMetric;
-import com.att.nsa.metrics.impl.CdmStringConstant;
-import com.att.nsa.metrics.impl.CdmTimeSince;
-
-/*@Component("dMaaPMetricsSet")*/
-/**
- * Metrics related information
- * 
- * @author anowarul.islam
- *
- */
-public class DMaaPMetricsSet extends CdmMetricsRegistryImpl implements MetricsSet {
-
-	private final CdmStringConstant fVersion;
-	private final CdmConstant fStartTime;
-	private final CdmTimeSince fUpTime;
-
-	private final CdmCounter fRecvTotal;
-	private final CdmRateTicker fRecvEpsInstant;
-	private final CdmRateTicker fRecvEpsShort;
-	private final CdmRateTicker fRecvEpsLong;
-
-	private final CdmCounter fSendTotal;
-	private final CdmRateTicker fSendEpsInstant;
-	private final CdmRateTicker fSendEpsShort;
-	private final CdmRateTicker fSendEpsLong;
-
-	private final CdmCounter fKafkaConsumerCacheMiss;
-	private final CdmCounter fKafkaConsumerCacheHit;
-
-	private final CdmCounter fKafkaConsumerClaimed;
-	private final CdmCounter fKafkaConsumerTimeout;
-
-	private final CdmSimpleMetric fFanOutRatio;
-
-	private final HashMap<String, CdmRateTicker> fPathUseRates;
-	private final HashMap<String, CdmMovingAverage> fPathAvgs;
-
-	private rrNvReadable fSettings;
-
-	private final ScheduledExecutorService fScheduler;
-
-	/**
-	 * Constructor initialization
-	 * 
-	 * @param cs
-	 */
-	
-		public DMaaPMetricsSet(rrNvReadable cs) {
-		
-		fVersion = new CdmStringConstant("Version " + CambriaApiVersionInfo.getVersion());
-		super.putItem("version", fVersion);
-
-		final long startTime = System.currentTimeMillis();
-		final Date d = new Date(startTime);
-		final String text = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssz").format(d);
-		fStartTime = new CdmConstant(startTime / 1000, "Start Time (epoch); " + text);
-		super.putItem("startTime", fStartTime);
-
-		fUpTime = new CdmTimeSince("seconds since start");
-		super.putItem("upTime", fUpTime);
-
-		fRecvTotal = new CdmCounter("Total events received since start");
-		super.putItem("recvTotalEvents", fRecvTotal);
-
-		fRecvEpsInstant = new CdmRateTicker("recv eps (1 min)", 1, TimeUnit.SECONDS, 1, TimeUnit.MINUTES);
-		super.putItem("recvEpsInstant", fRecvEpsInstant);
-
-		fRecvEpsShort = new CdmRateTicker("recv eps (10 mins)", 1, TimeUnit.SECONDS, 10, TimeUnit.MINUTES);
-		super.putItem("recvEpsShort", fRecvEpsShort);
-
-		fRecvEpsLong = new CdmRateTicker("recv eps (1 hr)", 1, TimeUnit.SECONDS, 1, TimeUnit.HOURS);
-		super.putItem("recvEpsLong", fRecvEpsLong);
-
-		fSendTotal = new CdmCounter("Total events sent since start");
-		super.putItem("sendTotalEvents", fSendTotal);
-
-		fSendEpsInstant = new CdmRateTicker("send eps (1 min)", 1, TimeUnit.SECONDS, 1, TimeUnit.MINUTES);
-		super.putItem("sendEpsInstant", fSendEpsInstant);
-
-		fSendEpsShort = new CdmRateTicker("send eps (10 mins)", 1, TimeUnit.SECONDS, 10, TimeUnit.MINUTES);
-		super.putItem("sendEpsShort", fSendEpsShort);
-
-		fSendEpsLong = new CdmRateTicker("send eps (1 hr)", 1, TimeUnit.SECONDS, 1, TimeUnit.HOURS);
-		super.putItem("sendEpsLong", fSendEpsLong);
-
-		fKafkaConsumerCacheMiss = new CdmCounter("Kafka Consumer Cache Misses");
-		super.putItem("kafkaConsumerCacheMiss", fKafkaConsumerCacheMiss);
-
-		fKafkaConsumerCacheHit = new CdmCounter("Kafka Consumer Cache Hits");
-		super.putItem("kafkaConsumerCacheHit", fKafkaConsumerCacheHit);
-
-		fKafkaConsumerClaimed = new CdmCounter("Kafka Consumers Claimed");
-		super.putItem("kafkaConsumerClaims", fKafkaConsumerClaimed);
-
-		fKafkaConsumerTimeout = new CdmCounter("Kafka Consumers Timedout");
-		super.putItem("kafkaConsumerTimeouts", fKafkaConsumerTimeout);
-
-		// FIXME: CdmLevel is not exactly a great choice
-		fFanOutRatio = new CdmSimpleMetric() {
-			@Override
-			public String getRawValueString() {
-				return getRawValue().toString();
-			}
-
-			@Override
-			public Number getRawValue() {
-				final double s = fSendTotal.getValue();
-				final double r = fRecvTotal.getValue();
-				return r == 0.0 ? 0.0 : s / r;
-			}
-
-			@Override
-			public String summarize() {
-				return getRawValueString() + " sends per recv";
-			}
-
-		};
-		super.putItem("fanOut", fFanOutRatio);
-
-		// these are added to the metrics catalog as they're discovered
-		fPathUseRates = new HashMap<String, CdmRateTicker>();
-		fPathAvgs = new HashMap<String, CdmMovingAverage>();
-
-		fScheduler = Executors.newScheduledThreadPool(1);
-	}
-
-	@Override
-	public void setupCambriaSender() {
-		DMaaPMetricsSender.sendPeriodically(fScheduler, this,  "cambria.apinode.metrics.dmaap");
-	}
-
-	@Override
-	public void onRouteComplete(String name, long durationMs) {
-		CdmRateTicker ticker = fPathUseRates.get(name);
-		if (ticker == null) {
-			ticker = new CdmRateTicker("calls/min on path " + name + "", 1, TimeUnit.MINUTES, 1, TimeUnit.HOURS);
-			fPathUseRates.put(name, ticker);
-			super.putItem("pathUse_" + name, ticker);
-		}
-		ticker.tick();
-
-		CdmMovingAverage durs = fPathAvgs.get(name);
-		if (durs == null) {
-			durs = new CdmMovingAverage("ms avg duration on path " + name + ", last 10 minutes", 10, TimeUnit.MINUTES);
-			fPathAvgs.put(name, durs);
-			super.putItem("pathDurationMs_" + name, durs);
-		}
-		durs.tick(durationMs);
-	}
-
-	@Override
-	public void publishTick(int amount) {
-		if (amount > 0) {
-			fRecvTotal.bumpBy(amount);
-			fRecvEpsInstant.tick(amount);
-			fRecvEpsShort.tick(amount);
-			fRecvEpsLong.tick(amount);
-		}
-	}
-
-	@Override
-	public void consumeTick(int amount) {
-		if (amount > 0) {
-			fSendTotal.bumpBy(amount);
-			fSendEpsInstant.tick(amount);
-			fSendEpsShort.tick(amount);
-			fSendEpsLong.tick(amount);
-		}
-	}
-
-	@Override
-	public void onKafkaConsumerCacheMiss() {
-		fKafkaConsumerCacheMiss.bump();
-	}
-
-	@Override
-	public void onKafkaConsumerCacheHit() {
-		fKafkaConsumerCacheHit.bump();
-	}
-
-	@Override
-	public void onKafkaConsumerClaimed() {
-		fKafkaConsumerClaimed.bump();
-	}
-
-	@Override
-	public void onKafkaConsumerTimeout() {
-		fKafkaConsumerTimeout.bump();
-	}
-
-}
diff --git a/src/main/java/com/att/dmf/mr/beans/DMaaPNsaApiDb.java b/src/main/java/com/att/dmf/mr/beans/DMaaPNsaApiDb.java
deleted file mode 100644
index 963ff2d..0000000
--- a/src/main/java/com/att/dmf/mr/beans/DMaaPNsaApiDb.java
+++ /dev/null
@@ -1,140 +0,0 @@
-/*******************************************************************************
- *  ============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.dmf.mr.beans;
-
-import java.security.Key;
-
-
-import org.springframework.beans.factory.annotation.Autowired;
-
-import com.att.dmf.mr.constants.CambriaConstants;
-import com.att.eelf.configuration.EELFLogger;
-import com.att.eelf.configuration.EELFManager;
-import com.att.nsa.configs.ConfigDb;
-import com.att.nsa.configs.ConfigDbException;
-import com.att.nsa.configs.confimpl.EncryptingLayer;
-import com.att.nsa.drumlin.till.nv.rrNvReadable;
-import com.att.nsa.drumlin.till.nv.rrNvReadable.missingReqdSetting;
-import com.att.nsa.security.db.BaseNsaApiDbImpl;
-import com.att.nsa.security.db.EncryptingApiDbImpl;
-import com.att.nsa.security.db.NsaApiDb;
-import com.att.nsa.security.db.simple.NsaSimpleApiKey;
-import com.att.nsa.security.db.simple.NsaSimpleApiKeyFactory;
-import com.att.nsa.util.rrConvertor;
-
-/**
- * 
- * @author anowarul.islam
- *
- */
-public class DMaaPNsaApiDb {
-	
-	
-	private DMaaPZkConfigDb cdb;
-	
-	//private static final Logger log = Logger
-		
-	private static final EELFLogger log = EELFManager.getInstance().getLogger(DMaaPNsaApiDb.class);
-	
-/**
- * 
- * Constructor initialized
- * @param settings
- * @param cdb
- */
-	@Autowired
-	public DMaaPNsaApiDb(rrNvReadable settings, DMaaPZkConfigDb cdb) {
-		
-		this.setCdb(cdb);
-	}
-	/**
-	 * 
-	 * @param settings
-	 * @param cdb
-	 * @return
-	 * @throws ConfigDbException
-	 * @throws missingReqdSetting
-	 */
-	public static NsaApiDb<NsaSimpleApiKey> buildApiKeyDb(
-			rrNvReadable settings, ConfigDb cdb) throws ConfigDbException,
-			missingReqdSetting {
-		// Cambria uses an encrypted api key db
-
-		
-		final String keyBase64 =com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,"cambria.secureConfig.key");
-		
-		
-	
-	final String initVectorBase64 =com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,"cambria.secureConfig.iv");
-		// if neither value was provided, don't encrypt api key db
-		if (keyBase64 == null && initVectorBase64 == null) {
-			log.info("This server is configured to use an unencrypted API key database. See the settings documentation.");
-			return new BaseNsaApiDbImpl<>(cdb,
-					new NsaSimpleApiKeyFactory());
-		} else if (keyBase64 == null) {
-			// neither or both, otherwise something's goofed
-			throw new missingReqdSetting("cambria.secureConfig.key");
-		} else if (initVectorBase64 == null) {
-			// neither or both, otherwise something's goofed
-			throw new missingReqdSetting("cambria.secureConfig.iv");
-		} else {
-			log.info("This server is configured to use an encrypted API key database.");
-			final Key key = EncryptingLayer.readSecretKey(keyBase64);
-			final byte[] iv = rrConvertor.base64Decode(initVectorBase64);
-			return new EncryptingApiDbImpl<>(cdb,
-					new NsaSimpleApiKeyFactory(), key, iv);
-		}
-	}
-
-	/**
-	 * @return
-	 * returns settings
-	 */
-
-		
-	
-
-	/**
-	 * @param settings
-	 * set settings
-	 */
-	
-		
-	
-
-	 /**
-	 * @return
-	 * returns cbd
-	 */
-	public DMaaPZkConfigDb getCdb() {
-		return cdb;
-	}
-	/**
-	 * @param cdb
-	 * set cdb
-	 */
-	public void setCdb(DMaaPZkConfigDb cdb) {
-		this.cdb = cdb;
-	}
-
-
-}
diff --git a/src/main/java/com/att/dmf/mr/beans/DMaaPZkClient.java b/src/main/java/com/att/dmf/mr/beans/DMaaPZkClient.java
deleted file mode 100644
index 78a7426..0000000
--- a/src/main/java/com/att/dmf/mr/beans/DMaaPZkClient.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*******************************************************************************
- *  ============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.dmf.mr.beans;
-
-import org.I0Itec.zkclient.ZkClient;
-import org.springframework.beans.factory.annotation.Qualifier;
-
-import com.att.dmf.mr.utils.ConfigurationReader;
-import com.att.nsa.drumlin.till.nv.rrNvReadable;
-
-/**
- * Created for Zookeeper client which will read configuration and settings parameter
- * @author nilanjana.maity
- *
- */
-public class DMaaPZkClient extends ZkClient {
-
-	/**
-	 * This constructor will get the settings value from rrNvReadable
-     * and ConfigurationReader's zookeeper connection
-	 * @param settings
-	 */
-	public DMaaPZkClient(@Qualifier("propertyReader") rrNvReadable settings) {
-		super(ConfigurationReader.getMainZookeeperConnectionString());
-	}
-}
diff --git a/src/main/java/com/att/dmf/mr/beans/DMaaPZkConfigDb.java b/src/main/java/com/att/dmf/mr/beans/DMaaPZkConfigDb.java
deleted file mode 100644
index 5aa25fa..0000000
--- a/src/main/java/com/att/dmf/mr/beans/DMaaPZkConfigDb.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*******************************************************************************
- *  ============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.dmf.mr.beans;
-
-import org.springframework.beans.factory.annotation.Qualifier;
-
-import com.att.dmf.mr.utils.ConfigurationReader;
-import com.att.nsa.configs.confimpl.ZkConfigDb;
-import com.att.nsa.drumlin.till.nv.rrNvReadable;
-
-/**
- * Provide the zookeeper config db connection 
- * @author nilanjana.maity
- *
- */
-public class DMaaPZkConfigDb extends ZkConfigDb {
-	/**
-	 * This Constructor will provide the configuration details from the property reader
-     * and DMaaPZkClient
-	 * @param zk
-	 * @param settings
-	 */
-	public DMaaPZkConfigDb(@Qualifier("dMaaPZkClient") DMaaPZkClient zk,
-			@Qualifier("propertyReader") rrNvReadable settings) {
-		
-		
-		super(ConfigurationReader.getMainZookeeperConnectionString(),ConfigurationReader.getMainZookeeperConnectionSRoot());
-		
-	}
-	
-	
-}
diff --git a/src/main/java/com/att/dmf/mr/beans/LogDetails.java b/src/main/java/com/att/dmf/mr/beans/LogDetails.java
deleted file mode 100644
index b7fb325..0000000
--- a/src/main/java/com/att/dmf/mr/beans/LogDetails.java
+++ /dev/null
@@ -1,214 +0,0 @@
-/**
- * 
- */
-/*******************************************************************************
- *  ============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.dmf.mr.beans;
-
-import java.util.Date;
-
-import com.att.dmf.mr.constants.CambriaConstants;
-import com.att.dmf.mr.utils.Utils;
-
-/**
- * @author muzainulhaque.qazi
- *
- */
-
-public class LogDetails {
-	
-	private String publisherId;
-	private String topicId;
-	private String subscriberGroupId;
-	private String subscriberId;
-	private String publisherIp;
-	private String messageBatchId;
-	private String messageSequence;
-	private String messageTimestamp;
-	private String consumeTimestamp;
-	private String transactionIdTs;	
-	private String serverIp;
-	
-	private long messageLengthInBytes; 
-	private long totalMessageCount;
-	
-	private boolean transactionEnabled;
-	/**
-	 * This is for transaction enabled logging details
-	 *
-	 */
-	public LogDetails() {
-		super();
-	}
-
-	public String getTransactionId() {
-		StringBuilder transactionId = new StringBuilder();
-		transactionId.append(transactionIdTs);
-		transactionId.append(CambriaConstants.TRANSACTION_ID_SEPARATOR);
-		transactionId.append(publisherIp);
-		transactionId.append(CambriaConstants.TRANSACTION_ID_SEPARATOR);
-		transactionId.append(messageBatchId);
-		transactionId.append(CambriaConstants.TRANSACTION_ID_SEPARATOR);
-		transactionId.append(messageSequence);
-
-		return transactionId.toString();
-	}
-
-	public String getPublisherId() {
-		return publisherId;
-	}
-
-	public void setPublisherId(String publisherId) {
-		this.publisherId = publisherId;
-	}
-
-	public String getTopicId() {
-		return topicId;
-	}
-
-	public void setTopicId(String topicId) {
-		this.topicId = topicId;
-	}
-
-	public String getSubscriberGroupId() {
-		return subscriberGroupId;
-	}
-
-	public void setSubscriberGroupId(String subscriberGroupId) {
-		this.subscriberGroupId = subscriberGroupId;
-	}
-
-	public String getSubscriberId() {
-		return subscriberId;
-	}
-
-	public void setSubscriberId(String subscriberId) {
-		this.subscriberId = subscriberId;
-	}
-
-	public String getPublisherIp() {
-		return publisherIp;
-	}
-
-	public void setPublisherIp(String publisherIp) {
-		this.publisherIp = publisherIp;
-	}
-
-	public String getMessageBatchId() {
-		return messageBatchId;
-	}
-
-	public void setMessageBatchId(Long messageBatchId) {
-		this.messageBatchId = Utils.getFromattedBatchSequenceId(messageBatchId);
-	}
-
-	public String getMessageSequence() {
-		return messageSequence;
-	}
-
-	public void setMessageSequence(String messageSequence) {
-		this.messageSequence = messageSequence;
-	}
-
-	public String getMessageTimestamp() {
-		return messageTimestamp;
-	}
-
-	public void setMessageTimestamp(String messageTimestamp) {
-		this.messageTimestamp = messageTimestamp;
-	}
-
-	public String getPublishTimestamp() {
-		return Utils.getFormattedDate(new Date());
-	}
-
-	public String getConsumeTimestamp() {
-		return consumeTimestamp;
-	}
-
-	public void setConsumeTimestamp(String consumeTimestamp) {
-		this.consumeTimestamp = consumeTimestamp;
-	}
-
-	public long getMessageLengthInBytes() {
-		return messageLengthInBytes;
-	}
-
-	public void setMessageLengthInBytes(long messageLengthInBytes) {
-		this.messageLengthInBytes = messageLengthInBytes;
-	}
-
-	public long getTotalMessageCount() {
-		return totalMessageCount;
-	}
-
-	public void setTotalMessageCount(long totalMessageCount) {
-		this.totalMessageCount = totalMessageCount;
-	}
-
-	public boolean isTransactionEnabled() {
-		return transactionEnabled;
-	}
-
-	public void setTransactionEnabled(boolean transactionEnabled) {
-		this.transactionEnabled = transactionEnabled;
-	}
-
-	public String getTransactionIdTs() {
-		return transactionIdTs;
-	}
-
-	public void setTransactionIdTs(String transactionIdTs) {
-		this.transactionIdTs = transactionIdTs;
-	}
-
-	public String getPublisherLogDetails() {
-		
-			StringBuilder buffer = new StringBuilder();
-			buffer.append("[publisherId=" + publisherId);
-			buffer.append(", topicId=" + topicId);
-			buffer.append(", messageTimestamp=" + messageTimestamp);
-			buffer.append(", publisherIp=" + publisherIp);
-			buffer.append(", messageBatchId=" + messageBatchId);
-			buffer.append(", messageSequence=" + messageSequence );
-			buffer.append(", messageLengthInBytes=" + messageLengthInBytes);
-			buffer.append(", transactionEnabled=" + transactionEnabled);
-			buffer.append(", transactionId=" + getTransactionId());
-			buffer.append(", publishTimestamp=" + getPublishTimestamp());		
-			buffer.append(", serverIp=" + getServerIp()+"]");
-		return buffer.toString();
-		
-	}
-
-	public String getServerIp() {
-		return serverIp;
-	}
-
-	public void setServerIp(String serverIp) {
-		this.serverIp = serverIp;
-	}
-
-	public void setMessageBatchId(String messageBatchId) {
-		this.messageBatchId = messageBatchId;
-	}
-	
-}
diff --git a/src/main/java/com/att/dmf/mr/beans/TopicBean.java b/src/main/java/com/att/dmf/mr/beans/TopicBean.java
deleted file mode 100644
index a397921..0000000
--- a/src/main/java/com/att/dmf/mr/beans/TopicBean.java
+++ /dev/null
@@ -1,155 +0,0 @@
-/**
- * 
- */
-/*******************************************************************************
- *  ============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.dmf.mr.beans;
-
-import java.io.Serializable;
-
-import javax.xml.bind.annotation.XmlRootElement;
-
-/**
- * @author muzainulhaque.qazi
- *
- */
-@XmlRootElement
-public class TopicBean implements Serializable {
-
-	private static final long serialVersionUID = -8620390377775457949L;
-	private String topicName;
-	private String topicDescription;
-
-	private int partitionCount;
-	private int replicationCount;
-
-	private boolean transactionEnabled;
-
-	/**
-	 * constructor
-	 */
-	public TopicBean() {
-		super();
-	}
-
-	/**
-	 * constructor initialization with topic details name, description,
-	 * partition, replication, transaction
-	 * 
-	 * @param topicName
-	 * @param description
-	 * @param partitionCount
-	 * @param replicationCount
-	 * @param transactionEnabled
-	 */
-	public TopicBean(String topicName, String topicDescription, int partitionCount, int replicationCount,
-			boolean transactionEnabled) {
-		super();
-		this.topicName = topicName;
-		this.topicDescription = topicDescription;
-		this.partitionCount = partitionCount;
-		this.replicationCount = replicationCount;
-		this.transactionEnabled = transactionEnabled;
-	}
-
-	/**
-	 * @return
-	 * returns topic name which is of String type
-	 */
-	public String getTopicName() {
-		return topicName;
-	}
-
-	/**
-	 * @param topicName
-	 * set topic name  
-	 */
-	public void setTopicName(String topicName) {
-		this.topicName = topicName;
-	}
-
-
-	/**
-	 * @return
-	 * returns partition count which is of int type
-	 */
-	public int getPartitionCount() {
-		return partitionCount;
-	}
-
-	/**
-	 * @param partitionCount
-	 * set partition Count 
-	 */
-	public void setPartitionCount(int partitionCount) {
-		this.partitionCount = partitionCount;
-	}
-	
-	/**
-	 * @return
-	 * returns replication count which is of int type
-	 */
-	public int getReplicationCount() {
-		return replicationCount;
-	}
-	
-	/**
-	 * @param
-	 * set replication count which is of int type
-	 */
-	public void setReplicationCount(int replicationCount) {
-		this.replicationCount = replicationCount;
-	}
-	
-	/**
-	 * @return
-	 * returns boolean value which indicates whether transaction is Enabled 
-	 */
-	public boolean isTransactionEnabled() {
-		return transactionEnabled;
-	}
-	
-	/**
-	 * @param
-	 * sets boolean value which indicates whether transaction is Enabled 
-	 */
-	public void setTransactionEnabled(boolean transactionEnabled) {
-		this.transactionEnabled = transactionEnabled;
-	}
-
-	/**
-	 * 
-	 * @return returns description which is of String type
-	 */
-	public String getTopicDescription() {
-		return topicDescription;
-	}
-	/**
-	 * 
-	 * @param topicDescription
-	 * set description which is of String type
-	 */
-	public void setTopicDescription(String topicDescription) {
-		this.topicDescription = topicDescription;
-	}
-
-}