<modelVersion>4.0.0</modelVersion>
<groupId>org.onap.dmaap.messagerouter.msgrtr</groupId>
<artifactId>msgrtr</artifactId>
- <version>1.1.5-SNAPSHOT</version>
+ <version>1.1.7-SNAPSHOT</version>
<packaging>jar</packaging>
<name>dmaap-messagerouter-msgrtr</name>
<description>Message Router - Restful interface built for kafka</description>
<dependency>
<groupId>org.codehaus.groovy</groupId>
<artifactId>groovy-all</artifactId>
- <version>2.4.4</version>
+ <version>2.4.8</version>
<scope>compile</scope>
<exclusions>
<exclusion>
<artifactId>cxf-rt-transports-http-jetty</artifactId> <version>${cxf.version}</version>
</dependency> -->
<!-- Begin - Spring Dependencies for DI -->
- <dependency>
- <groupId>org.springframework</groupId>
- <artifactId>spring-core</artifactId>
- <version>${spring.version}</version>
- </dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-expression</artifactId>
- <version>4.3.16.RELEASE</version>
+ <version>4.3.18.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-webmvc</artifactId>
- <version>4.3.15.RELEASE</version>
+ <version>4.3.18.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
- <version>4.3.15.RELEASE</version>
+ <version>4.3.18.RELEASE</version>
</dependency>
<dependency>
<groupId>javax.servlet</groupId>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
- <version>0.11.0.0</version>
+ <version>0.11.0.3</version>
</dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ <version>23.6.1-jre</version>
+ </dependency>
<dependency>
<groupId>com.att.eelf</groupId>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
- <version>2.6.0</version>
+ <version>4.0.1</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-test</artifactId>
- <version>2.6.0</version>
+ <version>4.0.1</version>
</dependency>
public class CambriaApiException extends NsaAppException
{
+ /*
+ * defined long type constant serialVersionUID
+ */
+ private static final long serialVersionUID = 1L;
- private ErrorResponse errRes;
+ private transient ErrorResponse errRes;
/**
* Implements constructor CambriaApiException
* @param jsonObject
this.errRes = errRes;
}
- /*
- * defined long type constant serialVersionUID
- */
- private static final long serialVersionUID = 1L;
public ErrorResponse getErrRes() {
return errRes;
}
*******************************************************************************/
package com.att.dmf.mr.backends;
-import java.util.ArrayList;
/**
* A consumer interface. Consumers pull the next message from a given topic.
import com.att.dmf.mr.constants.CambriaConstants;
import com.att.dmf.mr.exception.DMaaPErrorMessages;
import com.att.dmf.mr.utils.ConfigurationReader;
-//import org.slf4j.Logger;
-//import org.slf4j.LoggerFactory;
+
+
import com.att.eelf.configuration.EELFLogger;
import com.att.eelf.configuration.EELFManager;
import com.att.nsa.metrics.CdmTimer;
// the server at least every 30 seconds, timing out after 2 minutes should
// be okay.
// FIXME: consider allowing the client to specify its expected call rate?
- private static final long kDefault_MustTouchEveryMs = 1000 * 60 * 2;
+ private static final long kDefault_MustTouchEveryMs = 1000L*60*2;
// check for expirations pretty regularly
private static final long kDefault_SweepEverySeconds = 15;
NOT_STARTED, CONNECTED, DISCONNECTED, SUSPENDED
}
- // @Qualifier("kafkalockavoid")
-
- // @Resource
- // @Qualifier("kafkalockavoid")
- // KafkaLiveLockAvoider2 kafkaLiveLockAvoider;
+
+
@Autowired
private DMaaPErrorMessages errorMessages;
- // KafkaLiveLockAvoider kafkaLiveLockAvoider = new KafkaLiveLockAvoider();
+
/**
* User defined exception class for kafka consumer cache
*
EnsurePath ensurePath = new EnsurePath(fBaseZkPath);
ensurePath.ensure(curator.getZookeeperClient());
- // final long freq = fSettings.getLong(kSetting_SweepEverySeconds,
- // kDefault_SweepEverySeconds);
+
+
long freq = kDefault_SweepEverySeconds;
String strkSetting_SweepEverySeconds = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
kSetting_SweepEverySeconds);
try {
curator.blockUntilConnected();
} catch (InterruptedException e) {
- // Ignore
- log.error("error while setting curator framework :" + e.getMessage());
+ log.error("error while setting curator framework :",e);
+ Thread.currentThread().interrupt();
}
}
if (getStatus() != KafkaConsumerCache.Status.CONNECTED)
throw new KafkaConsumerCacheException("The cache service is unavailable.");
ArrayList<Kafka011Consumer> kcl = new ArrayList<>();
- // final String consumerKey = makeConsumerKey(topic, consumerGroupId,
- // clientId);
+
+
Enumeration<String> strEnum = fConsumers.keys();
String consumerLocalKey = null;
while (strEnum.hasMoreElements()) {
if (consumerLocalKey.startsWith(topicgroup) && (!consumerLocalKey.endsWith("::" + clientId))) {
- // System.out.println("consumer key returning from
- // getConsumerListforCG +++++++++ " + consumerLocalKey
- // + " " + fConsumers.get(consumerLocalKey));
+
+
+
kcl.add(fConsumers.get(consumerLocalKey));
}
if (getStatus() != KafkaConsumerCache.Status.CONNECTED)
throw new KafkaConsumerCacheException("The cache service is unavailable.");
ArrayList<Kafka011Consumer> kcl = new ArrayList<>();
- // final String consumerKey = makeConsumerKey(topic, consumerGroupId,
- // clientId);
+
Enumeration<String> strEnum = fConsumers.keys();
String consumerLocalKey = null;
while (strEnum.hasMoreElements()) {
if (consumerLocalKey.startsWith(group)) {
- // System.out.println("consumer key returning from
- // getConsumerListforCG +++++++++ " + consumerLocalKey
- // + " " + fConsumers.get(consumerLocalKey));
+
kcl.add(fConsumers.get(consumerLocalKey));
}
final String consumerKey = makeConsumerKey(topic, consumerGroupId, consumerId);
fConsumers.put(consumerKey, consumer);
- // String appId = "node-instance-"+i;
+
log.info("^@ Consumer Added to Cache Consumer Key" + consumerKey + " ApiId" + fApiId);
}
consumerHandoverWaitMs = Integer.parseInt(strkSetting_ConsumerHandoverWaitMs);
Thread.sleep(consumerHandoverWaitMs);
} catch (InterruptedException e) {
- // Ignore
+ log.error("InterruptedException in dropTimedOutConsumer",e);
+ Thread.currentThread().interrupt();
}
log.info("Dropped " + key + " consumer due to timeout");
}
final Kafka011Consumer kc = fConsumers.get(key);
log.info("closing Kafka consumer " + key + " object " + kc);
if (kc != null) {
- // log.info("closing Kafka consumer " + key);
+
if (kc.close()) {
fConsumers.remove(key);
throws KafkaConsumerCacheException {
// get a lock at <base>/<topic>::<consumerGroupId>::<consumerId>
final String consumerKey = makeConsumerKey(topic, consumerGroupId, consumerId);
- final CdmTimer timer = new CdmTimer(fMetrics, "CacheSignalOwnership");
- try {
+ try(final CdmTimer timer = new CdmTimer(fMetrics, "CacheSignalOwnership")) {
final String consumerPath = fBaseZkPath + "/" + consumerKey;
log.debug(fApiId + " attempting to claim ownership of consumer " + consumerKey);
final CuratorFramework curator = ConfigurationReader.getCurator();
consumerHandoverWaitMs = Integer.parseInt(strkSetting_ConsumerHandoverWaitMs);
Thread.sleep(consumerHandoverWaitMs);
} catch (InterruptedException e) {
- // Ignore
+ log.error("InterruptedException in signalOwnership",e);
+ Thread.currentThread().interrupt();
}
}
mustTouchEveryMs = Long.parseLong(strkSetting_TouchEveryMs);
}
- // final long mustTouchEveryMs =
- // fSettings.getLong(kSetting_TouchEveryMs, kDefault_MustTouchEveryMs);
+
final long oldestAllowedTouchMs = System.currentTimeMillis() - mustTouchEveryMs;
for (Entry<String, Kafka011Consumer> e : fConsumers.entrySet()) {
}
private static final EELFLogger log = EELFManager.getInstance().getLogger(KafkaConsumerCache.class);
- // private static final Logger log =
- // LoggerFactory.getLogger(KafkaConsumerCache.class);
+
}
\ No newline at end of file
import com.att.eelf.configuration.EELFManager;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.context.annotation.ComponentScan;
import org.springframework.stereotype.Component;
-import org.springframework.stereotype.Service;
//@ComponentScan(basePackages="com.att.dmf.mr.backends.kafka")
@Component
@PostConstruct
public void init() {
- System.out.println("Welcome......................................................................................");
+ log.info("Welcome......................................................................................");
try {
if (curatorFramework.checkExists().forPath(locksPath) == null) {
curatorFramework.create().creatingParentsIfNeeded().forPath(locksPath);
}
} catch (Exception e) {
- //e.printStackTrace();
+
log.error("Error during creation of permanent Znodes under /live-lock-avoid ",e);
}
protected void assignNewProcessNode(String appId, Watcher processNodeWatcher ) {
String taskHolderZnodePath = ZNODE_ROOT+ZNODE_UNSTICK_TASKS+"/"+appId;
- //Watcher processNodeWatcher = createWatcher();
+
try {
import com.att.eelf.configuration.EELFManager;
import com.att.nsa.drumlin.till.nv.rrNvReadable;
-//import kafka.FailedToSendMessageException;
-//import kafka.javaapi.producer.Producer;
-//import kafka.producer.KeyedMessage;
-//import kafka.producer.ProducerConfig;
-//import kafka.producer.KeyedMessage;
+
/**
* Sends raw JSON objects into Kafka.
kafkaConnUrl="localhost:9092";
}
- //String jaaspath="C:/ATT/Apps/dmaapCodedmaap-framework/dmaap/bundleconfig-local/etc/appprops/kafka_pub_jaas.conf";
- // props.put("bootstrap.servers", bootSever);
- //System.setProperty("java.security.auth.login.config",jaaspath);
+
- /*transferSetting( props, "sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='admin_secret';");
- transferSetting( props, "security.protocol", "SASL_PLAINTEXT");
- transferSetting( props, "sasl.mechanism", "PLAIN");*/
+
+
transferSetting( props, "bootstrap.servers",kafkaConnUrl);
- //transferSetting( props, "metadata.broker.list", kafkaConnUrl);
+
transferSetting( props, "request.required.acks", "1");
transferSetting( props, "message.send.max.retries", "5");
transferSetting(props, "retry.backoff.ms", "150");
- //props.put("serializer.class", "kafka.serializer.StringEncoder");
+
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
- //fConfig = new ProducerConfig(props);
- //fProducer = new Producer<String, String>(fConfig);
+
+
fProducer = new KafkaProducer<>(props);
}
throws IOException {
log.info("sending " + msgs.size() + " events to [" + topic + "]");
try{
- final List<ProducerRecord<String, String>> kms = new ArrayList<ProducerRecord<String, String>>(msgs.size());
+ final List<ProducerRecord<String, String>> kms = new ArrayList<>(msgs.size());
for (message o : msgs) {
- final ProducerRecord<String, String> data = new ProducerRecord<String, String>(topic, o.getKey(), o.toString());
- //kms.add(data);
+ final ProducerRecord<String, String> data = new ProducerRecord<>(topic, o.getKey(), o.toString());
+
try {
}
//private final rrNvReadable fSettings;
- //private ProducerConfig fConfig;
+
private Producer<String, String> fProducer;
/**
}
- //@Override
- //public void sendBatchMessage(String topic, ArrayList<KeyedMessage<String, String>> kms) throws IOException {
- // TODO Auto-generated method stub
-
- //}
+
}
\ No newline at end of file
*/
public class MemoryConsumerFactory implements ConsumerFactory
{
+
+ private final MemoryQueue fQueue;
+
/**
*
* Initializing constructor
return new MemoryConsumer ( topic, consumerGroupId );
}
- private final MemoryQueue fQueue;
-
/**
*
* Define nested inner class
*/
private class MemoryConsumer implements Consumer
{
+
+ private final String fTopic;
+ private final String fConsumer;
+ private final long fCreateMs;
+ private long fLastAccessMs;
+
/**
*
* Initializing MemoryConsumer constructor
return fQueue.get ( fTopic, fConsumer );
}
- private final String fTopic;
- private final String fConsumer;
- private final long fCreateMs;
- private long fLastAccessMs;
-
@Override
public boolean close() {
//Nothing to close/clean up.
*/
public Collection<? extends Consumer> getConsumers ()
{
- return new ArrayList<MemoryConsumer> ();
+ return new ArrayList<> ();
}
@Override
*
*/
public class MemoryMetaBroker implements Broker {
+
+ private final MemoryQueue fQueue;
+ private final HashMap<String, MemTopic> fTopics;
+
/**
*
* @param mq
* @param settings
*/
public MemoryMetaBroker(MemoryQueue mq, ConfigDb configDb) {
- //public MemoryMetaBroker(MemoryQueue mq, ConfigDb configDb, rrNvReadable settings) {
+
fQueue = mq;
- fTopics = new HashMap<String, MemTopic>();
+ fTopics = new HashMap<>();
}
@Override
fQueue.removeTopic(topic);
}
- private final MemoryQueue fQueue;
- private final HashMap<String, MemTopic> fTopics;
-
private static class MemTopic implements Topic {
+
+ private final String fName;
+ private final String fDesc;
+ private final String fOwner;
+ private NsaAcl fReaders;
+ private NsaAcl fWriters;
+ private boolean ftransactionEnabled;
+ private String accessDenied = "User does not own this topic ";
+
/**
* constructor initialization
*
@Override
public void permitWritesFromUser(String publisherId, NsaApiKey asUser) throws AccessDeniedException {
if (!fOwner.equals(asUser.getKey())) {
- throw new AccessDeniedException("User does not own this topic " + fName);
+ throw new AccessDeniedException(accessDenied + fName);
}
if (fWriters == null) {
fWriters = new NsaAcl();
@Override
public void denyWritesFromUser(String publisherId, NsaApiKey asUser) throws AccessDeniedException {
if (!fOwner.equals(asUser.getKey())) {
- throw new AccessDeniedException("User does not own this topic " + fName);
+ throw new AccessDeniedException(accessDenied + fName);
}
fWriters.remove(publisherId);
}
@Override
public void permitReadsByUser(String consumerId, NsaApiKey asUser) throws AccessDeniedException {
if (!fOwner.equals(asUser.getKey())) {
- throw new AccessDeniedException("User does not own this topic " + fName);
+ throw new AccessDeniedException(accessDenied + fName);
}
if (fReaders == null) {
fReaders = new NsaAcl();
@Override
public void denyReadsByUser(String consumerId, NsaApiKey asUser) throws AccessDeniedException {
if (!fOwner.equals(asUser.getKey())) {
- throw new AccessDeniedException("User does not own this topic " + fName);
+ throw new AccessDeniedException(accessDenied + fName);
}
fReaders.remove(consumerId);
}
- private final String fName;
- private final String fDesc;
- private final String fOwner;
- private NsaAcl fReaders;
- private NsaAcl fWriters;
- private boolean ftransactionEnabled;
-
@Override
public boolean isTransactionEnabled() {
return ftransactionEnabled;
@Override
public Set<String> getOwners() {
- final TreeSet<String> set = new TreeSet<String> ();
+ final TreeSet<String> set = new TreeSet<> ();
set.add ( fOwner );
return set;
}
* constructor storing hashMap objects in Queue and Offsets object
*/
public MemoryQueue() {
- fQueue = new HashMap<String, LogBuffer>();
+ fQueue = new HashMap<>();
fOffsets = new HashMap<String, HashMap<String, Integer>>();
}
HashMap<String, Integer> offsetMap = fOffsets.get(consumerName);
if (offsetMap == null) {
- offsetMap = new HashMap<String, Integer>();
+ offsetMap = new HashMap<>();
fOffsets.put(consumerName, offsetMap);
}
Integer offset = offsetMap.get(topic);
public LogBuffer(int maxSize) {
fBaseOffset = 0;
fMaxSize = maxSize;
- fList = new ArrayList<String>();
+ fList = new ArrayList<>();
}
/**
import com.att.dmf.mr.exception.ErrorResponse;
import com.att.dmf.mr.utils.Utils;
-//import org.slf4j.Logger;
-//import org.slf4j.LoggerFactory;
-
import com.att.eelf.configuration.EELFLogger;
import com.att.eelf.configuration.EELFManager;
import com.att.nsa.drumlin.service.standards.HttpStatusCodes;
*/
@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
*
* @throws invalidSettingValue
*/
@Autowired
- public DMaaPCambriaLimiter(@Qualifier("propertyReader") rrNvReadable settings)
- throws missingReqdSetting, invalidSettingValue {
- fRateInfo = new HashMap<String, RateInfo>();
- fRateInfoCheck = new HashMap<String, RateInfoCheck>();
+ 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,
5000);
}
-
- /**
- * 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));
- }
-
+
/**
* Construct a rate limiter.
*
* @param windowLengthMins
*/
public DMaaPCambriaLimiter(double maxEmptyPollsPerMinute,double maxPollsPerMinute, int windowLengthMins, long sleepMs ,long sleepMS1) {
- fRateInfo = new HashMap<String, RateInfo>();
- fRateInfoCheck = new HashMap<String, RateInfoCheck>();
+ fRateInfo = new HashMap<>();
+ fRateInfoCheck = new HashMap<>();
fMaxEmptyPollsPerMinute = Math.max(0, maxEmptyPollsPerMinute);
fMaxPollsPerMinute = Math.max(0, maxPollsPerMinute);
fWindowLengthMins = windowLengthMins;
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.
log.info(ri.getLabel() + ": " + "No sleep configured, just throwing error.");
}
} catch (InterruptedException e) {
+ log.error("Exception "+ e);
// ignore
}
log.info(errRes.toString());
throw new CambriaApiException(errRes);
}
- /*if (fMaxPollsPerMinute <= 0) {
- return;
- }
- final RateInfoCheck ric = getRateInfoCheck(topic, consumerGroup, clientId);
- final double ratevalue = ric.onCall();
- if (ratevalue > fMaxPollsPerMinute) {
- try {
- log.warn(ri.getLabel() + ": " + rate + " empty replies/minute, limit is " + fMaxEmptyPollsPerMinute
- + ".");
- if (fSleepMs1 > fMaxPollsPerMinute) {
- log.warn(ri.getLabel() + ": " + "Slowing response with " + fSleepMs
- + " ms sleep, then responding in error.");
- Thread.sleep(fSleepMs1);
- ric.reset();
- } else {
- log.info(ri.getLabel() + ": " + "No sleep configured, just throwing error.");
- }
- } catch (InterruptedException e) {
- // ignore
- }
-
-
- ErrorResponse errRes = new ErrorResponse(HttpStatusCodes.k429_tooManyRequests,
- DMaaPResponseCode.TOO_MANY_REQUESTS.getResponseCode(),
- "This client is making too many requests "
- + ",decrease the number of requests. ","",Utils.getFormattedDate(new Date()),topic,"","",consumerGroup+"/"+clientId,remoteHost);
-
- log.info(errRes.toString());
- throw new CambriaApiException(errRes);
- }*/
}
}
private static class RateInfo {
+ private final String fLabel;
+ private final CdmRateTicker fCallRateSinceLastMsgSend;
/**
* constructor initialzes
*
fCallRateSinceLastMsgSend.tick();
return fCallRateSinceLastMsgSend.getRate();
}
-
- private final String fLabel;
- private final CdmRateTicker fCallRateSinceLastMsgSend;
}
private static class RateInfoCheck {
+
+ private final String fLabel;
+ private final CdmRateTicker fCallRateSinceLastMsgSend;
/**
* constructor initialzes
*
fCallRateSinceLastMsgSend.tick();
return fCallRateSinceLastMsgSend.getRate();
}
-
- private final String fLabel;
- private final CdmRateTicker fCallRateSinceLastMsgSend;
}
- 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 Logger log = LoggerFactory.getLogger(DMaaPCambriaLimiter.class);
- private static final EELFLogger log = EELFManager.getInstance().getLogger(DMaaPCambriaLimiter.class);
+
private RateInfo getRateInfo(String topic, String consumerGroup, String clientId) {
final String key = makeKey(topic, consumerGroup, clientId);
}
- private RateInfoCheck getRateInfoCheck(String topic, String consumerGroup, String clientId) {
- final String key = makeKey(topic, consumerGroup, clientId);
- RateInfoCheck ri = fRateInfoCheck.get(key);
- if (ri == null) {
- ri = new RateInfoCheck(key, 1);
- fRateInfoCheck.put(key, ri);
- }
- return ri;
- }
+
import com.att.dmf.mr.backends.kafka.LiveLockAvoidance;
import com.att.dmf.mr.constants.CambriaConstants;
import com.att.dmf.mr.utils.ConfigurationReader;
-//import org.slf4j.Logger;
-//import org.slf4j.LoggerFactory;
+
+
import com.att.eelf.configuration.EELFLogger;
import com.att.eelf.configuration.EELFManager;
import com.att.nsa.drumlin.till.nv.rrNvReadable.missingReqdSetting;
*/
public class DMaaPKafkaConsumerFactory implements ConsumerFactory {
- // private static final Logger log = LoggerFactory
- // .getLogger(DMaaPKafkaConsumerFactory.class);
+
private static final EELFLogger log = EELFManager.getInstance().getLogger(DMaaPKafkaConsumerFactory.class);
- // @Autowired
- // private KafkaLiveLockAvoider kafkaLiveLockAvoider = new
- // KafkaLiveLockAvoider();
+
/**
* constructor initialization
final boolean isCacheEnabled = kSetting_EnableCache;
- // fCache = (isCacheEnabled) ? new KafkaConsumerCache(apiNodeId,
- // metrics) : null;
+
fCache = null;
if (isCacheEnabled) {
fCache = KafkaConsumerCache.getInstance();
log.info("Creating Kafka consumer for group [" + consumerGroupName + "], consumer [" + consumerId
+ "], on topic [" + topic + "].");
-
- fCache.signalOwnership(topic, consumerGroupName, consumerId);
-
+
+ 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);// ,fCache.getkafkaLiveLockAvoiderObj()
- // );
+ kc = new Kafka011Consumer(topic, consumerGroupName, consumerId, cc, fkafkaLiveLockAvoider);
log.info(" kafka stream created in " + (System.currentTimeMillis() - fCreateTimeMs));
if (fCache != null) {
private void transferSettingIfProvided(Properties target, String key, String prefix) {
String keyVal = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, makeLongKey(key, prefix));
- // if (fSettings.hasValueFor(makeLongKey(key, prefix))) {
+
if (null != keyVal) {
- // final String val = fSettings
- // .getString(makeLongKey(key, prefix), "");
+
log.info("Setting [" + key + "] to " + keyVal + ".");
target.put(key, keyVal);
}
props.put("group.id", fakeGroupName);
props.put("enable.auto.commit", "false"); // 0.11
props.put("bootstrap.servers", fkafkaBrokers);
- /*props.put("sasl.jaas.config",
- "org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='admin_secret';");
- 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
//import org.apache.log4-j.Logger;
import com.att.eelf.configuration.EELFLogger;
import com.att.eelf.configuration.EELFManager;
-//import com.att.dmf.mr.backends.kafka.kafka011.SettingsUtil;
+
import com.att.nsa.configs.ConfigDb;
import com.att.nsa.configs.ConfigDbException;
import com.att.nsa.configs.ConfigPath;
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, fkafkaBrokers );
- /* props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='admin_secret';");
- props.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
- props.put("sasl.mechanism", "PLAIN");*/
+
fKafkaAdminClient=AdminClient.create ( props );
- // fKafkaAdminClient = null;
+
}
//private static final Logger log = Logger.getLogger(DMaaPKafkaMetaBroker.class);
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, fkafkaBrokers );
- /* props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='admin_secret';");
- props.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
- props.put("sasl.mechanism", "PLAIN");*/
+
fKafkaAdminClient=AdminClient.create ( props );
- // fKafkaAdminClient = null;
+
}
public DMaaPKafkaMetaBroker( rrNvReadable settings,
ZkClient zk, ConfigDb configDb,AdminClient client) {
- //fSettings = settings;
+
fZk = zk;
fCambriaConfig = configDb;
fBaseTopicData = configDb.parse("/topics");
fKafkaAdminClient= client;
- // fKafkaAdminClient = null;
+
}
}
catch ( InterruptedException e )
{
- //timer.fail ( "Timeout" );
+
log.warn ( "Execution of describeTopics timed out." );
throw new ConfigDbException ( e );
}
catch ( ExecutionException e )
{
- //timer.fail ( "ExecutionError" );
+
log.warn ( "Execution of describeTopics failed: " + e.getCause ().getMessage (), e.getCause () );
throw new ConfigDbException ( e.getCause () );
}
log.info("Loading zookeeper client for topic deletion.");
// topic creation. (Otherwise, the topic is only partially created
// in ZK.)
- /*zkClient = new ZkClient(ConfigurationReader.getMainZookeeperConnectionString(), 10000, 10000,
- ZKStringSerializer$.MODULE$);
- String strkSettings_KafkaZookeeper = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,CambriaConstants.kSetting_ZkConfigDbServers);
- if (null==strkSettings_KafkaZookeeper) strkSettings_KafkaZookeeper = CambriaConstants.kDefault_ZkConfigDbServers;
- ZkUtils zkutils =new ZkUtils(zkClient , new ZkConnection(strkSettings_KafkaZookeeper),false);
- */
+
fKafkaAdminClient.deleteTopics(Arrays.asList(topic));
log.info("Zookeeper client loaded successfully. Deleting topic.");
- //AdminUtils.deleteTopic(zkutils, topic);
+
} catch (Exception e) {
log.error("Failed to delete topic [" + topic + "]. " + e.getMessage(), e);
throw new ConfigDbException(e);
*
* @param cs
*/
- //public DMaaPMetricsSet() {
+
public DMaaPMetricsSet(rrNvReadable cs) {
- //fSettings = cs;
+
fVersion = new CdmStringConstant("Version " + CambriaApiVersionInfo.getVersion());
super.putItem("version", fVersion);
import java.security.Key;
-//import org.apache.log4-j.Logger;
+
import org.springframework.beans.factory.annotation.Autowired;
import com.att.dmf.mr.constants.CambriaConstants;
*/
public class DMaaPNsaApiDb {
- //private rrNvReadable settings;
+
private DMaaPZkConfigDb cdb;
//private static final Logger log = Logger
- // .getLogger(DMaaPNsaApiDb.class.toString());
+
private static final EELFLogger log = EELFManager.getInstance().getLogger(DMaaPNsaApiDb.class);
/**
*/
@Autowired
public DMaaPNsaApiDb(rrNvReadable settings, DMaaPZkConfigDb cdb) {
- //this.setSettings(settings);
+
this.setCdb(cdb);
}
/**
missingReqdSetting {
// Cambria uses an encrypted api key db
- //final String keyBase64 = settings.getString("cambria.secureConfig.key", null);
+
final String keyBase64 =com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,"cambria.secureConfig.key");
- // final String initVectorBase64 = settings.getString( "cambria.secureConfig.iv", null);
+
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<NsaSimpleApiKey>(cdb,
+ return new BaseNsaApiDbImpl<>(cdb,
new NsaSimpleApiKeyFactory());
} else if (keyBase64 == null) {
// neither or both, otherwise something's goofed
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<NsaSimpleApiKey>(cdb,
+ return new EncryptingApiDbImpl<>(cdb,
new NsaSimpleApiKeyFactory(), key, iv);
}
}
* @return
* returns settings
*/
-/* public rrNvReadable getSettings() {
- return settings;
- }*/
+
+
+
/**
* @param settings
* set settings
*/
- /*public void setSettings(rrNvReadable settings) {
- this.settings = settings;
- }*/
+
+
+
/**
* @return
import com.att.dmf.mr.utils.ConfigurationReader;
import com.att.nsa.configs.confimpl.ZkConfigDb;
import com.att.nsa.drumlin.till.nv.rrNvReadable;
-//import com.att.nsa.configs.confimpl.ZkConfigDb;
+
/**
* Provide the zookeeper config db connection
* @author nilanjana.maity
public DMaaPZkConfigDb(@Qualifier("dMaaPZkClient") DMaaPZkClient zk,
@Qualifier("propertyReader") rrNvReadable settings) {
- //super(com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,CambriaConstants.kSetting_ZkConfigDbRoot)==null?CambriaConstants.kDefault_ZkConfigDbRoot:com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,CambriaConstants.kSetting_ZkConfigDbRoot));
+
super(ConfigurationReader.getMainZookeeperConnectionString(),ConfigurationReader.getMainZookeeperConnectionSRoot());
}
import javax.ws.rs.ext.Provider;
import org.apache.http.HttpStatus;
-//import org.apache.log-4j.Logger;
+
import org.springframework.beans.factory.annotation.Autowired;
import com.att.eelf.configuration.EELFLogger;
@Singleton
public class DMaaPWebExceptionMapper implements ExceptionMapper<WebApplicationException>{
- //private static final Logger LOGGER = Logger
- // .getLogger(DMaaPWebExceptionMapper.class);
+
private static final EELFLogger LOGGER = EELFManager.getInstance().getLogger(DMaaPWebExceptionMapper.class);
private ErrorResponse errRes;
public class CambriaServletContextListener implements ServletContextListener {
DME2EndPointLoader loader = DME2EndPointLoader.getInstance();
-// private static Logger log = Logger.getLogger(CambriaServletContextListener.class);
+
private static final EELFLogger log = EELFManager.getInstance().getLogger(CambriaServletContextListener.class);
private String protocol;
private String serviceURL;
private static DME2EndPointLoader loader = new DME2EndPointLoader();
-// private static final Logger LOG = LoggerFactory.getLogger(EventsServiceImpl.class);
+
private static final EELFLogger LOG = EELFManager.getInstance().getLogger(EventsServiceImpl.class);
private DME2EndPointLoader() {
}
*
*//*
public class AccessDeniedException extends Exception
- {
+
*//**
* AccessDenied Description
*//*
- public AccessDeniedException () { super ( "Access denied." ); }
+
*//**
* AccessDenied Exception for the user while authenticating the user request
* @param user
*//*
- public AccessDeniedException ( String user ) { super ( "Access denied for " + user ); }
+
private static final long serialVersionUID = 1L;
}*/
*******************************************************************************/
package com.att.dmf.mr.metrics.publisher;
-//import org.slf4j.Logger;
+
//
import com.att.eelf.configuration.EELFLogger;
-//import com.att.eelf.configuration.EELFManager;
+
/**
*
*/
public static List<HttpHost> createHostsList(Collection<String> hosts)
{
- final ArrayList<HttpHost> convertedHosts = new ArrayList<HttpHost> ();
+ final ArrayList<HttpHost> convertedHosts = new ArrayList<>();
for ( String host : hosts )
{
if ( host.length () == 0 ) continue;
* Your API secret
* @return an identity manager
*/
- /*
- * public static CambriaIdentityManager createIdentityManager (
- * Collection<String> hostSet, String apiKey, String apiSecret ) { final
- * CambriaIdentityManager cim = new CambriaMetaClient ( hostSet );
- * cim.setApiCredentials ( apiKey, apiSecret ); return cim; }
- */
+
/**
* Create a topic manager for working with topics.
* Your API secret
* @return a topic manager
*/
- /*
- * public static CambriaTopicManager createTopicManager ( Collection<String>
- * hostSet, String apiKey, String apiSecret ) { final CambriaMetaClient tmi
- * = new CambriaMetaClient ( hostSet ); tmi.setApiCredentials ( apiKey,
- * apiSecret ); return tmi; }
- */
+
/**
* Inject a consumer. Used to support unit tests.
import org.json.JSONException;
import com.att.dmf.mr.constants.CambriaConstants;
-//import org.slf4j.Logger;
+
//import org.slf4j.LoggerFactory;
import com.att.eelf.configuration.EELFLogger;
import com.att.eelf.configuration.EELFManager;
public CambriaBaseClient ( Collection<String> hosts, String clientSignature ) throws MalformedURLException
{
- /*super ( hosts, CambriaConstants.kStdCambriaServicePort, clientSignature,
- CacheUse.NONE, 1, 1, TimeUnit.MILLISECONDS );*/
+
+
super(ConnectionType.HTTP, hosts, CambriaConstants.kStdCambriaServicePort, clientSignature, CacheUse.NONE, 1, 1L, TimeUnit.MILLISECONDS, 32, 32, 600000);
- //fLog = LoggerFactory.getLogger ( this.getClass().getName () );
+
fLog = EELFManager.getInstance().getLogger(this.getClass().getName());
//( this.getClass().getName () );
}
{
fLog = log;
- //replaceLogger ( log );
+
}
public EELFLogger getLog ()
public void close() {
try {
final List<message> remains = close(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
- if (remains.size() > 0) {
+ if (remains.isEmpty()) {
getLog().warn("Closing publisher with " + remains.size() + " messages unsent. "
+ "Consider using CambriaBatchingPublisher.close( long timeout, TimeUnit timeoutUnits ) to recapture unsent messages on close.");
}
*/
private synchronized boolean shouldSendNow() {
boolean shouldSend = false;
- if (fPending.size() > 0) {
+ if (fPending.isEmpty()) {
final long nowMs = Clock.now();
shouldSend = (fPending.size() >= fMaxBatchSize);
private synchronized boolean sendBatch() {
// it's possible for this call to be made with an empty list. in this
// case, just return.
- if (fPending.size() < 1) {
+ if (fPending.isEmpty()) {
return true;
}
// code from REST Client Starts
- // final String serverCalculatedSignature = sha1HmacSigner.sign
- // ("2015-09-21T11:38:19-0700", "iHAxArrj6Ve9JgmHvR077QiV");
+
+
Client client = ClientBuilder.newClient();
String metricTopicname = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,"metrics.send.cambria.topic");
Entity<byte[]> data = Entity.entity(baseStream.toByteArray(), "application/cambria");
Response response = target.request().post(data);
- // header("X-CambriaAuth",
- // "2OH46YIWa329QpEF:"+serverCalculatedSignature).
- // header("X-CambriaDate", "2015-09-21T11:38:19-0700").
- // post(Entity.json(baseStream.toByteArray()));
-
+
getLog().info("Response received :: " + response.getStatus());
getLog().info("Response received :: " + response.toString());
// code from REST Client Ends
- /*
- * final JSONObject result = post ( url, contentType,
- * baseStream.toByteArray(), true ); final String logLine =
- * "cambria reply ok (" + (Clock.now()-startMs) + " ms):" +
- * result.toString (); getLog().info ( logLine );
- */
+
fPending.clear();
return true;
} catch (IllegalArgumentException x) {
getLog().warn(x.getMessage(), x);
}
- /*
- * catch ( HttpObjectNotFoundException x ) { getLog().warn (
- * x.getMessage(), x ); } catch ( HttpException x ) { getLog().warn (
- * x.getMessage(), x ); }
- */
+
catch (IOException x) {
getLog().warn(x.getMessage(), x);
}
final int c = fTokens.next();
- /*if (c ==','){
- fCloseCount++;
- System.out.println("fCloseCount=" + fCloseCount +" fCount "+fCount);
- }*/
+
if (fIsList) {
if (c == ']' || (fCount > 0 && c == 10))
return null;
*
* @param o
*/
- //public msg(JSONObject o){}
+
public msg(JSONObject o) {
private final InputStream fStream;
private final String fDefPart;
private boolean fClosed;
- //private String transactionId;
+
}
auth = true;
}
- //System.out.println("role " +role +" user: "+ req.getRemoteUser() +" : auth="+auth);
+
return auth;
}
String permission = "";
String nameSpace ="";
if(topicName.contains(".") && topicName.contains("com.att")) {
- //String topic = topicName.substring(topicName.lastIndexOf(".")+1);
+
nameSpace = topicName.substring(0,topicName.lastIndexOf("."));
}
else {
if(null==nameSpace)nameSpace="com.att.dmaap.mr.ueb";
- /*ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN,
- DMaaPResponseCode.TOPIC_NOT_IN_AAF.getResponseCode(), "Topic does not exist in AAF"
- , null, Utils.getFormattedDate(new Date()), topicName,
- null, null, null, null);
-
- throw new CambriaApiException(errRes);*/
+
}
permission = nameSpace+".mr.topic|:topic."+topicName+"|"+action;
import com.att.dmf.mr.beans.DMaaPContext;
import com.att.dmf.mr.security.DMaaPAuthenticator;
-//import com.att.nsa.security.db.NsaApiDb;
+
import com.att.eelf.configuration.EELFLogger;
import com.att.eelf.configuration.EELFManager;
import com.att.nsa.security.NsaApiKey;
log.info ( "AUTH-LOG(" + remoteAddr + "): " + msg );
}
-// private final NsaApiDb<K> fDb;
+
//private static final Logger log = Logger.getLogger( MechIdAuthenticator.class.toString());
private static final EELFLogger log = EELFManager.getInstance().getLogger(MechIdAuthenticator.class);
/**
public DMaaPOriginalUebAuthenticator(NsaApiDb<K> db, long requestTimeWindowMs) {
fDb = db;
fRequestTimeWindowMs = requestTimeWindowMs;
- //fAuthenticators = new LinkedList<DMaaPAuthenticator<K>>();
+
- //fAuthenticators.add(new DMaaPOriginalUebAuthenticator<K>(db, requestTimeWindowMs));
+
}
"EEEE, dd-MMM-yy HH:mm:ss zzz",
};
- /*private static final String kDateFormats[] = {
- // W3C date format (RFC 3339).
- "yyyy-MM-dd'T'HH:mm:ssz",
+
+
+
- // Preferred HTTP date format (RFC 1123).
- "EEE, dd MMM yyyy HH:mm:ss zzz",
+
+
- // simple unix command line 'date' format
- "EEE MMM dd HH:mm:ss z yyyy",
+
+
- // Common date format (RFC 822).
- "EEE, dd MMM yy HH:mm:ss z", "EEE, dd MMM yy HH:mm z", "dd MMM yy HH:mm:ss z", "dd MMM yy HH:mm z",
+
+
- // Obsoleted HTTP date format (ANSI C asctime() format).
- "EEE MMM dd HH:mm:ss yyyy",
+
+
- // Obsoleted HTTP date format (RFC 1036).
- "EEEE, dd-MMM-yy HH:mm:ss zzz", }; */
+
+
// logger declaration
- //private static final Logger log = Logger.getLogger(DMaaPOriginalUebAuthenticator.class.toString());
+
private static final EELFLogger log = EELFManager.getInstance().getLogger(DMaaPOriginalUebAuthenticator.class);
@Override
-// public K authenticate(DMaaPContext ctx) {
+
// TODO Auto-generated method stub
- //return null;
+
//}
public K authenticate(DMaaPContext ctx) {
- /*final HttpServletRequest req = ctx.getRequest();
- for (DMaaPAuthenticator<K> a : fAuthenticators) {
- if (a.qualify(req)) {
- final K k = a.isAuthentic(req);
- if (k != null)
- return k;
- }
- // else: this request doesn't look right to the authenticator
- }*/
+
+
+
+
+
+
+
+
+
return null;
}
public void addAuthenticator ( DMaaPAuthenticator<K> a )
{
- //this.fAuthenticators.add(a);
+
}
- //private final LinkedList<DMaaPAuthenticator<K>> fAuthenticators;
+
}
\ No newline at end of file
import com.att.nsa.limits.Blacklist;
import com.att.nsa.security.NsaApiKey;
import com.att.nsa.security.ReadWriteSecuredResource.AccessDeniedException;
-//import com.att.sa.highlandPark.util.HpJsonUtil;
+
/**
* @author muzainulhaque.qazi
@Service
public class ApiKeysServiceImpl implements ApiKeysService {
- //private Logger log = Logger.getLogger(ApiKeysServiceImpl.class.toString());
+
private static final EELFLogger log = EELFManager.getInstance().getLogger(ApiKeysServiceImpl.class.toString());
/**
* This method will provide all the ApiKeys present in kafka server.
String kSetting_AllowAnonymousKeys= com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,"apiKeys.allowAnonymous");
if(null==kSetting_AllowAnonymousKeys) kSetting_AllowAnonymousKeys ="false";
- // if ((contactEmail == null) || (contactEmail.length() == 0))
+
if ( kSetting_AllowAnonymousKeys.equalsIgnoreCase("true") && !emailProvided )
{
DMaaPResponseBuilder.respondWithErrorInJson(dmaapContext, 400, "You must provide an email address.");
log.debug("=======ApiKeysServiceImpl: createApiKey : saving api key : "
+ key.toString() + "=====");
apiKeyDb.saveApiKey(key);
- // System.out.println("here4");
+
// email out the secret to validate the email address
if ( emailProvided )
{
);
DMaaPResponseBuilder.respondOk(dmaapContext,
o);
- /*o.put("secret", "Emailed to " + contactEmail + ".");
- DMaaPResponseBuilder.respondOk(dmaapContext,
- o); */
+
return;
} else {
log.debug("=======ApiKeysServiceImpl: createApiKey : Error in creating API Key.=====");
@Service
public class EventsServiceImpl implements EventsService {
// private static final Logger LOG =
- // Logger.getLogger(EventsServiceImpl.class);
+
private static final EELFLogger LOG = EELFManager.getInstance().getLogger(EventsServiceImpl.class);
private static final String BATCH_LENGTH = "event.batch.length";
private DMaaPErrorMessages errorMessages;
//@Autowired
- //KafkaLiveLockAvoider2 kafkaLiveLockAvoider;
+
// @Value("${metrics.send.cambria.topic}")
- // private String metricsTopic;
+
public DMaaPErrorMessages getErrorMessages() {
return errorMessages;
CambriaApiException, IOException, DMaaPAccessDeniedException {
final long startTime = System.currentTimeMillis();
final HttpServletRequest req = ctx.getRequest();
- //System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>"+kafkaLiveLockAvoider);
+
boolean isAAFTopic = false;
// was this host blacklisted?
final String remoteAddr = Utils.getRemoteAddress(ctx);
if (strtimeoutMS != null)
timeoutMs = Integer.parseInt(strtimeoutMS);
// int timeoutMs = ctx.getConfigReader().getSettings().getInt("timeout",
- // CambriaConstants.kNoTimeout);
+
if (req.getParameter("timeout") != null) {
timeoutMs = Integer.parseInt(req.getParameter("timeout"));
}
// if headers are not provided then user will be null
if (user == null && null != ctx.getRequest().getHeader("Authorization")) {
// the topic name will be sent by the client
- // String permission = "com.att.dmaap.mr.topic"+"|"+topic+"|"+"sub";
+
DMaaPAAFAuthenticator aaf = new DMaaPAAFAuthenticatorImpl();
String permission = aaf.aafPermissionString(topic, "sub");
if (!aaf.aafAuthentication(ctx.getRequest(), permission)) {
logger.info("Time taken in getEvents Authorization " + elapsedMs1 + " ms for " + topic + " " + consumerGroup
+ " " + clientId);
Consumer c = null;
- // String localclientId = clientId;
+
String lhostId = com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
"clusterhostid");
if (null == lhostId) {
// start processing, building a batch to push to the backend
final long startMs = System.currentTimeMillis();
long count = 0;
- long maxEventBatch = 1024 * 16;
+ long maxEventBatch = 1024L* 16;
String batchlen = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, BATCH_LENGTH);
if (null != batchlen)
maxEventBatch = Long.parseLong(batchlen);
// long maxEventBatch =
// ctx.getConfigReader().getSettings().getLong(BATCH_LENGTH, 1024 * 16);
- final LinkedList<Publisher.message> batch = new LinkedList<Publisher.message>();
+ final LinkedList<Publisher.message> batch = new LinkedList<>();
// final ArrayList<KeyedMessage<String, String>> kms = new
// ArrayList<KeyedMessage<String, String>>();
- final ArrayList<ProducerRecord<String, String>> pms = new ArrayList<ProducerRecord<String, String>>();
+ final ArrayList<ProducerRecord<String, String>> pms = new ArrayList<>();
try {
// for each message...
Publisher.message m = null;
// start processing, building a batch to push to the backend
final long startMs = System.currentTimeMillis();
long count = 0;
- long maxEventBatch = 1024 * 16;
+ long maxEventBatch = 1024L * 16;
String evenlen = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, BATCH_LENGTH);
if (null != evenlen)
maxEventBatch = Long.parseLong(evenlen);
@Component
public class MetricsServiceImpl implements MetricsService {
- //private static final Logger LOG = Logger.getLogger(MetricsService.class.toString());
+
private static final EELFLogger LOG = EELFManager.getInstance().getLogger(MetricsService.class);
/**
*
import com.att.dmf.mr.exception.ErrorResponse;
import com.att.dmf.mr.metabroker.Broker.TopicExistsException;
import com.att.dmf.mr.metabroker.Broker1;
-//import com.att.dmf.mr.metabroker.Broker1;
+
import com.att.dmf.mr.metabroker.Topic;
import com.att.dmf.mr.security.DMaaPAAFAuthenticator;
import com.att.dmf.mr.security.DMaaPAAFAuthenticatorImpl;
public class TopicServiceImpl implements TopicService {
// private static final Logger LOGGER =
- // Logger.getLogger(TopicServiceImpl.class);
+
private static final EELFLogger LOGGER = EELFManager.getInstance().getLogger(TopicServiceImpl.class);
@Autowired
private DMaaPErrorMessages errorMessages;
// @Value("${msgRtr.topicfactory.aaf}")
- // private String mrFactory;
+
public DMaaPErrorMessages getErrorMessages() {
return errorMessages;
for (Topic topic : getMetaBroker(dmaapContext).getAllTopics()) {
JSONObject obj = new JSONObject();
obj.put("topicName", topic.getName());
- // obj.put("description", topic.getDescription());
+
obj.put("owner", topic.getOwner());
obj.put("txenabled", topic.isTransactionEnabled());
topicsList.put(obj);
final NsaApiKey user = DMaaPAuthenticatorImpl.getAuthenticatedUser(dmaapContext);
String key = null;
- //String appName = dmaapContext.getRequest().getHeader("AppName");
+
String enfTopicName = com.att.ajsc.beans.PropertiesMapBean.getProperty(CambriaConstants.msgRtr_prop,
"enforced.topic.name.AAF");
"Failed to create topic: Access Denied.User does not have permission to perform create topic");
LOGGER.info(errRes.toString());
- // throw new DMaaPAccessDeniedException(errRes);
+
}
}
- // else if (user==null &&
+
// (null==dmaapContext.getRequest().getHeader("Authorization") && null
- // == dmaapContext.getRequest().getHeader("cookie")) ) {
- /*else if (user == null && null == dmaapContext.getRequest().getHeader("Authorization")
- ) {
- LOGGER.error("Failed to create topic" + topicBean.getTopicName() + ", Authentication failed.");
+
+
+
+
- ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_UNAUTHORIZED,
- DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(),
- "Failed to create topic: Access Denied.User does not have permission to perform create topic");
+
+
+
- LOGGER.info(errRes.toString());
- // throw new DMaaPAccessDeniedException(errRes);
- }*/
+
+
+
if (user == null /*&& (null != dmaapContext.getRequest().getHeader("Authorization")
)*/) {
- // if (user == null &&
+
// (null!=dmaapContext.getRequest().getHeader("Authorization") ||
- // null != dmaapContext.getRequest().getHeader("cookie"))) {
+
// ACL authentication is not provided so we will use the aaf
// authentication
- /*LOGGER.info("Authorization the topic");
+
- String permission = "";
- String nameSpace = "";
- if (topicBean.getTopicName().indexOf(".") > 1)
- nameSpace = topicBean.getTopicName().substring(0, topicBean.getTopicName().lastIndexOf("."));
+
+
+
+
- String mrFactoryVal = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
- "msgRtr.topicfactory.aaf");
+
+
- // AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,kSettings_KafkaZookeeper);
+
- permission = mrFactoryVal + nameSpace + "|create";
- DMaaPAAFAuthenticator aaf = new DMaaPAAFAuthenticatorImpl();*/
+
+
- //if (!aaf.aafAuthentication(dmaapContext.getRequest(), permission)) {
+
if (false) {
LOGGER.error("Failed to create topic" + topicBean.getTopicName() + ", Authentication failed.");
ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_UNAUTHORIZED,
DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(),
"Failed to create topic: Access Denied.User does not have permission to create topic with perm "
- //+ permission);
+
+ "permission");
} else {
// if user is null and aaf authentication is ok then key should
// be ""
- // key = "";
+
/**
* Added as part of AAF user it should return username
*/
- //key = dmaapContext.getRequest().getUserPrincipal().getName().toString();
- key="admin";
- LOGGER.info("key ==================== " + key);
+
+
+ //LOGGER.info("key ==================== " + key);
}
}
final String topicName = topicBean.getTopicName();
final String desc = topicBean.getTopicDescription();
int partition = topicBean.getPartitionCount();
- // int replica = topicBean.getReplicationCount();
+
if (partition == 0) {
partition = 8;
}
int replica = topicBean.getReplicationCount();
if (replica == 0) {
- //replica = 3;
+
replica = 1;
}
final int replicas = replica;
throw new CambriaApiException(errRes);
} catch (com.att.dmf.mr.metabroker.Broker1.TopicExistsException e) {
// TODO Auto-generated catch block
- e.printStackTrace();
+ LOGGER.error("Exception is at createTopic( ) ", e);
}
}
LOGGER.info("Granting write access to producer [" + producerId + "] for topic " + topicName);
final NsaApiKey user = DMaaPAuthenticatorImpl.getAuthenticatedUser(dmaapContext);
- // if (user == null) {
+
//
// LOGGER.info("Authenticating the user, as ACL authentication is not
- // provided");
+
//// String permission =
- // "com.att.dmaap.mr.topic"+"|"+topicName+"|"+"manage";
+
//
- // DMaaPAAFAuthenticator aaf = new DMaaPAAFAuthenticatorImpl();
- // String permission = aaf.aafPermissionString(topicName, "manage");
+
+
// if(!aaf.aafAuthentication(dmaapContext.getRequest(), permission))
// {
// LOGGER.error("Failed to permit write access to producer [" +
// producerId + "] for topic " + topicName
- // + ". Authentication failed.");
+
// ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN,
// DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(),
// errorMessages.getNotPermitted1()+" <Grant publish permissions>
- // "+errorMessages.getNotPermitted2()+ topicName);
- // LOGGER.info(errRes);
+
+
// throw new DMaaPAccessDeniedException(errRes);
// }
// }
LOGGER.info("Revoking write access to producer [" + producerId + "] for topic " + topicName);
final NsaApiKey user = DMaaPAuthenticatorImpl.getAuthenticatedUser(dmaapContext);
- // if (user == null) {
+
//
//// String permission =
- // "com.att.dmaap.mr.topic"+"|"+topicName+"|"+"manage";
+
// DMaaPAAFAuthenticator aaf = new DMaaPAAFAuthenticatorImpl();
// String permission = aaf.aafPermissionString(topicName, "manage");
// if(!aaf.aafAuthentication(dmaapContext.getRequest(), permission))
// {
// LOGGER.error("Failed to revoke write access to producer [" +
// producerId + "] for topic " + topicName
- // + ". Authentication failed.");
+
// ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN,
// DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(),
// errorMessages.getNotPermitted1()+" <Revoke publish permissions>
- // "+errorMessages.getNotPermitted2()+ topicName);
- // LOGGER.info(errRes);
+
+
// throw new DMaaPAccessDeniedException(errRes);
//
- // }
+
// }
Topic topic = getMetaBroker(dmaapContext).getTopic(topicName);
LOGGER.info("Granting read access to consumer [" + consumerId + "] for topic " + topicName);
final NsaApiKey user = DMaaPAuthenticatorImpl.getAuthenticatedUser(dmaapContext);
- // if (user == null) {
+
//
//// String permission =
- // "com.att.dmaap.mr.topic"+"|"+topicName+"|"+"manage";
- // DMaaPAAFAuthenticator aaf = new DMaaPAAFAuthenticatorImpl();
+
+
// String permission = aaf.aafPermissionString(topicName, "manage");
// if(!aaf.aafAuthentication(dmaapContext.getRequest(), permission))
// {
// LOGGER.error("Failed to permit read access to consumer [" +
// consumerId + "] for topic " + topicName
- // + ". Authentication failed.");
+
// ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN,
// DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(),
// errorMessages.getNotPermitted1()+" <Grant consume permissions>
- // "+errorMessages.getNotPermitted2()+ topicName);
- // LOGGER.info(errRes);
+
+
// throw new DMaaPAccessDeniedException(errRes);
// }
// }
LOGGER.info("Revoking read access to consumer [" + consumerId + "] for topic " + topicName);
final NsaApiKey user = DMaaPAuthenticatorImpl.getAuthenticatedUser(dmaapContext);
- // if (user == null) {
+
//// String permission =
- // "com.att.dmaap.mr.topic"+"|"+topicName+"|"+"manage";
+
// DMaaPAAFAuthenticator aaf = new DMaaPAAFAuthenticatorImpl();
// String permission = aaf.aafPermissionString(topicName, "manage");
// if(!aaf.aafAuthentication(dmaapContext.getRequest(), permission))
// {
// LOGGER.error("Failed to revoke read access to consumer [" +
// consumerId + "] for topic " + topicName
- // + ". Authentication failed.");
+
// ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN,
// DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(),
// errorMessages.getNotPermitted1()+" <Grant consume permissions>
- // "+errorMessages.getNotPermitted2()+ topicName);
+
// LOGGER.info(errRes);
// throw new DMaaPAccessDeniedException(errRes);
// }
//
//
- // }
-
+
Topic topic = getMetaBroker(dmaapContext).getTopic(topicName);
if (null == topic) {
throws ConfigDbException, IOException {
/*
- * ConfigurationReader configReader = dmaapContext.getConfigReader();
+
*
* LOG.info("configReader : "+configReader.toString());
*
IOException {
/*
- * if (null != transactionId) {
+
*
* ConfigurationReader configReader = dmaapContext.getConfigReader();
*
import java.util.Map;
import java.util.Map.Entry;
-//import kafka.common.TopicExistsException;
+
import org.apache.kafka.common.errors.TopicExistsException;
import org.json.JSONArray;
import org.json.JSONObject;
@Service
public class UIServiceImpl implements UIService {
- //private static final Logger LOGGER = Logger.getLogger(UIServiceImpl.class);
+
private static final EELFLogger LOGGER = EELFManager.getInstance().getLogger(UIServiceImpl.class);
/**
* Returning template of hello page
import com.att.nsa.drumlin.till.nv.rrNvReadable.missingReqdSetting;
import com.att.nsa.limits.Blacklist;
import com.att.nsa.security.NsaAuthenticatorService;
-//import com.att.nsa.security.authenticators.OriginalUebAuthenticator;
+
import com.att.nsa.security.db.BaseNsaApiDbImpl;
import com.att.nsa.security.db.NsaApiDb;
import com.att.nsa.security.db.NsaApiDb.KeyExistsException;
@Component
public class ConfigurationReader {
-// private rrNvReadable settings;
+
private Broker1 fMetaBroker;
private ConsumerFactory fConsumerFactory;
private Publisher fPublisher;
@Autowired
private DMaaPCambriaLimiter fRateLimiter;
private NsaApiDb<NsaSimpleApiKey> fApiKeyDb;
- /* private DMaaPTransactionObjDB<DMaaPTransactionObj> fTranDb; */
+
private DMaaPAuthenticator<NsaSimpleApiKey> fSecurityManager;
private NsaAuthenticatorService<NsaSimpleApiKey> nsaSecurityManager;
private static CuratorFramework curator;
private Emailer fEmailer;
private static final EELFLogger log = EELFManager.getInstance().getLogger(ConfigurationReader.class);
- //private static final Logger log = Logger.getLogger(ConfigurationReader.class.toString());
+
/**
* constructor to initialize all the values
@Qualifier("dMaaPAuthenticatorImpl") DMaaPAuthenticator<NsaSimpleApiKey> fSecurityManager
)
throws missingReqdSetting, invalidSettingValue, ServletException, KafkaConsumerCacheException, ConfigDbException {
- //this.settings = settings;
+
this.fMetrics = fMetrics;
this.zk = zk;
this.fConfigDb = fConfigDb;
ConfigurationReader.curator = curator;
this.fConsumerFactory = fConsumerFactory;
this.fMetaBroker = fMetaBroker;
- //System.out.println("SSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSs " + fMetaBroker);
+
this.q = q;
this.mmb = mmb;
this.fApiKeyDb = fApiKeyDb;
- /* this.fTranDb = fTranDb; */
+
this.fSecurityManager = fSecurityManager;
long allowedtimeSkewMs=600000L;
String strallowedTimeSkewM= com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,"authentication.allowedTimeSkewMs");
if(null!=strallowedTimeSkewM)allowedtimeSkewMs= Long.parseLong(strallowedTimeSkewM);
- // boolean requireSecureChannel = true;
+
//String strrequireSecureChannel= com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,"aauthentication.requireSecureChannel");
//if(strrequireSecureChannel!=null)requireSecureChannel=Boolean.parseBoolean(strrequireSecureChannel);
//this.nsaSecurityManager = new NsaAuthenticatorService<NsaSimpleApiKey>(this.fApiKeyDb, settings.getLong("authentication.allowedTimeSkewMs", 600000L), settings.getBoolean("authentication.requireSecureChannel", true));
*/
public static void respondOkWithStream(DMaaPContext ctx, String mediaType, StreamWriter writer) throws IOException {
ctx.getResponse().setStatus(200);
- OutputStream os = getStreamForBinaryResponse(ctx, mediaType);
- writer.write(os);
- os.close();
-
+ try(OutputStream os = getStreamForBinaryResponse(ctx, mediaType)) {
+ writer.write(os);
+ }
+
}
/**
* interface used to define write method for outputStream
*/
- public static abstract interface StreamWriter {
+ public abstract static interface StreamWriter {
/**
* abstract method used to write the response
*
boolean fResponseEntityAllowed = (!(ctx.getRequest().getMethod().equalsIgnoreCase("HEAD")));
-
- OutputStream os = null;
- try{
+
if (fResponseEntityAllowed) {
- os = ctx.getResponse().getOutputStream();
- return os;
+ try(OutputStream os = ctx.getResponse().getOutputStream()){
+ return os;
+ }catch (Exception e){
+ log.error("Exception in getStreamForBinaryResponse",e);
+ throw new IOException();
+ }
} else {
- os = new NullStream();
- return os;
- }
- }catch (Exception e){
- throw new IOException();
-
- }
- finally{
- if(null != os){
- try{
- os.close();
- }catch(Exception e) {
-
- }
+ try(OutputStream os = new NullStream()){
+ return os;
+ }catch (Exception e){
+ log.error("Exception in getStreamForBinaryResponse",e);
+ throw new IOException();
}
}
}
* initializing logger
*
*/
- //private static final Logger LOGGER = Logger.getLogger(PropertyReader.class);
+
private static final EELFLogger log = EELFManager.getInstance().getLogger(PropertyReader.class);
-// private static final String MSGRTR_PROPERTIES_FILE = "msgRtrApi.properties";
+
/**
* constructor initialization
*
*/
public PropertyReader() throws loadException {
- /* Map<String, String> argMap = new HashMap<String, String>();
- final String config = getSetting(argMap, CambriaConstants.kConfig, MSGRTR_PROPERTIES_FILE);
- final URL settingStream = findStream(config, ConfigurationReader.class);
- push(new nvPropertiesFile(settingStream));
- push(new nvReadableTable(argMap));*/
+
+
+
+
+
}
/**
* @exception MalformedURLException
*
*/
- /*public static URL findStream(final String resourceName, Class<?> clazz) {
- try {
- File file = new File(resourceName);
+
+
+
- if (file.isAbsolute()) {
- return file.toURI().toURL();
- }
+
+
+
- String filesRoot = System.getProperty("RRWT_FILES", null);
+
- if (null != filesRoot) {
+
- String fullPath = filesRoot + "/" + resourceName;
+
- LOGGER.debug("Looking for [" + fullPath + "].");
+
- file = new File(fullPath);
- if (file.exists()) {
- return file.toURI().toURL();
- }
- }
+
+
+
+
+
- URL res = clazz.getClassLoader().getResource(resourceName);
+
- if (null != res) {
- return res;
- }
+
+
+
- res = ClassLoader.getSystemResource(resourceName);
+
+
+
+
+
+
+
+
+
+
- if (null != res) {
- return res;
- }
- } catch (MalformedURLException e) {
- LOGGER.error("Unexpected failure to convert a local filename into a URL: " + e.getMessage(), e);
- }
- return null;
- }
-*/
}
String Setting_CambriaTopic=com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,kSetting_CambriaTopic);
if(Setting_CambriaTopic==null) Setting_CambriaTopic = "msgrtr.apinode.metrics.dmaap";
- // Setting_CambriaBaseUrl=Setting_CambriaBaseUrl==null?defaultTopic:Setting_CambriaBaseUrl;
+
String Setting_CambriaSendFreqSecs=com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,kSetting_CambriaSendFreqSecs);
private final CambriaPublisher fCambria;
private final String fHostname;
- //private static final Logger log = LoggerFactory.getLogger(MetricsSender.class);
+
private static final EELFLogger log = EELFManager.getInstance().getLogger(MetricsSender.class);
/**
private FilterConfig filterConfig = null;
DMaaPErrorMessages errorMessages = null;
- //private Logger log = Logger.getLogger(ContentLengthFilter.class.toString());
+
private static final EELFLogger log = EELFManager.getInstance().getLogger(ContentLengthFilter.class);
/**
* Default constructor.
DMaaPResponseCode.MSG_SIZE_EXCEEDS_MSG_LIMIT.getResponseCode(), errorMessages.getMsgSizeExceeds()
+ jsonObj.toString());
log.info(errRes.toString());
- // throw new CambriaApiException(errRes);
+
}
}
\r
import org.apache.curator.CuratorZookeeperClient;\r
import org.apache.curator.framework.CuratorFramework;\r
+import org.apache.curator.framework.WatcherRemoveCuratorFramework;\r
import org.apache.curator.framework.api.CreateBuilder;\r
import org.apache.curator.framework.api.CuratorListener;\r
import org.apache.curator.framework.api.DeleteBuilder;\r
import org.apache.curator.framework.api.ExistsBuilder;\r
import org.apache.curator.framework.api.GetACLBuilder;\r
import org.apache.curator.framework.api.GetChildrenBuilder;\r
+import org.apache.curator.framework.api.GetConfigBuilder;\r
import org.apache.curator.framework.api.GetDataBuilder;\r
+import org.apache.curator.framework.api.ReconfigBuilder;\r
+import org.apache.curator.framework.api.RemoveWatchesBuilder;\r
import org.apache.curator.framework.api.SetACLBuilder;\r
import org.apache.curator.framework.api.SetDataBuilder;\r
import org.apache.curator.framework.api.SyncBuilder;\r
import org.apache.curator.framework.api.UnhandledErrorListener;\r
+import org.apache.curator.framework.api.transaction.CuratorMultiTransaction;\r
import org.apache.curator.framework.api.transaction.CuratorTransaction;\r
+import org.apache.curator.framework.api.transaction.TransactionOp;\r
import org.apache.curator.framework.imps.CuratorFrameworkState;\r
import org.apache.curator.framework.listen.Listenable;\r
+import org.apache.curator.framework.schema.SchemaSet;\r
+import org.apache.curator.framework.state.ConnectionStateErrorPolicy;\r
import org.apache.curator.framework.state.ConnectionStateListener;\r
import org.apache.curator.utils.EnsurePath;\r
import org.apache.zookeeper.Watcher;\r
+import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;\r
\r
public class CuratorFrameworkImpl implements CuratorFramework {\r
\r
return null;\r
}\r
\r
+ @Override\r
+ public ReconfigBuilder reconfig() {\r
+ // TODO Auto-generated method stub\r
+ return null;\r
+ }\r
+\r
+ @Override\r
+ public GetConfigBuilder getConfig() {\r
+ // TODO Auto-generated method stub\r
+ return null;\r
+ }\r
+\r
+ @Override\r
+ public CuratorMultiTransaction transaction() {\r
+ // TODO Auto-generated method stub\r
+ return null;\r
+ }\r
+\r
+ @Override\r
+ public TransactionOp transactionOp() {\r
+ // TODO Auto-generated method stub\r
+ return null;\r
+ }\r
+\r
+ @Override\r
+ public void createContainers(String path) throws Exception {\r
+ // TODO Auto-generated method stub\r
+ \r
+ }\r
+\r
+ @Override\r
+ public RemoveWatchesBuilder watches() {\r
+ // TODO Auto-generated method stub\r
+ return null;\r
+ }\r
+\r
+ @Override\r
+ public WatcherRemoveCuratorFramework newWatcherRemoveCuratorFramework() {\r
+ // TODO Auto-generated method stub\r
+ return null;\r
+ }\r
+\r
+ @Override\r
+ public ConnectionStateErrorPolicy getConnectionStateErrorPolicy() {\r
+ // TODO Auto-generated method stub\r
+ return null;\r
+ }\r
+\r
+ @Override\r
+ public QuorumVerifier getCurrentConfig() {\r
+ // TODO Auto-generated method stub\r
+ return null;\r
+ }\r
+\r
+ @Override\r
+ public SchemaSet getSchemaSet() {\r
+ // TODO Auto-generated method stub\r
+ return null;\r
+ }\r
+\r
+ @Override\r
+ public boolean isZk34CompatibilityMode() {\r
+ // TODO Auto-generated method stub\r
+ return false;\r
+ }\r
+\r
}\r
major=1
minor=1
-patch=5
+patch=7
base_version=${major}.${minor}.${patch}
Code Review / dmaap / messagerouter / msgrtr.git / commitdiff