* ONAP-Logging
* ================================================================================
* Copyright (C) 2017-2021 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2023 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
// --- init logger first
static {
- loggerType = initlogger();
+ loggerType = initLogger();
}
/**
* @param clazz the class
*/
public static Logger getLogger(Class<?> clazz) {
- Logger logger = null;
- displayMessage(GET_LOGGER_PREFIX + loggerType);
- switch (loggerType) {
-
- case EELF:
- logger = getEelfLogger(clazz, false);
- break;
- case SYSTEMOUT:
- default:
- logger = getSystemOutLogger();
- break;
- }
-
- return logger;
-
+ return getLogger(clazz, false);
}
/**
* Returns an instance of Logger.
*/
public static Logger getLogger() {
- Logger logger = null;
- displayMessage(GET_LOGGER_PREFIX + loggerType);
- switch (loggerType) {
-
- case EELF:
- logger = getEelfLogger(null, false);
- break;
- case SYSTEMOUT:
- default:
- logger = getSystemOutLogger();
- break;
- }
-
- return logger;
-
+ return getLogger(null);
}
/**
* Returns an instance of Logger.
*
- * @param clazz the class
+ * @param clazz the class
* @param isNewTransaction is a new transaction
*/
public static Logger getLogger(Class<?> clazz, boolean isNewTransaction) {
- Logger logger = null;
+ Logger logger;
displayMessage(GET_LOGGER_PREFIX + loggerType);
- switch (loggerType) {
-
- case EELF:
- logger = getEelfLogger(clazz, isNewTransaction);
- break;
- case SYSTEMOUT:
- default:
- logger = getSystemOutLogger();
- break;
+ if (loggerType == LoggerType.EELF) {
+ logger = getEelfLogger(clazz, isNewTransaction);
+ } else {
+ logger = getSystemOutLogger();
}
return logger;
-
}
/**
* @param isNewTransaction is a new transaction
*/
public static Logger getLogger(boolean isNewTransaction) {
- Logger logger = null;
- displayMessage(GET_LOGGER_PREFIX + loggerType);
- switch (loggerType) {
-
- case EELF:
- logger = getEelfLogger(null, isNewTransaction);
- break;
- case SYSTEMOUT:
- default:
- logger = getSystemOutLogger();
- break;
- }
-
- return logger;
+ return getLogger(null, isNewTransaction);
}
/**
/**
* Returns an instance of EelfLogger.
*
- * @param clazz the class
+ * @param clazz the class
* @param isNewTransaction is a new transaction
*/
private static EelfLogger getEelfLogger(Class<?> clazz, boolean isNewTransaction) {
/**
* loads the logger properties.
*/
- private static LoggerType initlogger() {
+ private static LoggerType initLogger() {
var loggerType = LoggerType.EELF;
Properties properties = null;
properties = PropertyUtil.getProperties("config/policyLogger.properties");
displayMessage("FlexLogger:properties => " + properties);
- if (properties != null) {
- String overrideLogbackLevel = properties.getProperty("override.logback.level.setup");
- displayMessage("FlexLogger:overrideLogbackLevel => " + overrideLogbackLevel);
- var loggerTypeString = properties.getProperty("logger.type");
- if ("EELF".equalsIgnoreCase(loggerTypeString) && "TRUE".equalsIgnoreCase(overrideLogbackLevel)) {
- displayMessage("FlexLogger: start listener.");
- properties = PropertyUtil.getProperties("config/policyLogger.properties",
- new PropertiesCallBack("FlexLogger-CallBack"));
- }
+ String overrideLogbackLevel = properties.getProperty("override.logback.level.setup");
+ displayMessage("FlexLogger:overrideLogbackLevel => " + overrideLogbackLevel);
+ var loggerTypeString = properties.getProperty("logger.type");
+ if ("EELF".equalsIgnoreCase(loggerTypeString) && "TRUE".equalsIgnoreCase(overrideLogbackLevel)) {
+ displayMessage("FlexLogger: start listener.");
+ properties = PropertyUtil.getProperties("config/policyLogger.properties",
+ new PropertiesCallBack("FlexLogger-CallBack"));
}
} catch (IOException e1) {
- displayMessage("initlogger" + e1);
+ displayMessage("initLogger" + e1);
} finally {
// OK to pass no properties (null)
loggerType = PolicyLogger.init(properties);
var sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS+00:00");
var startTime = Instant.now();
- String formatedTime = sdf.format(Date.from(startTime));
- displayMessage("FlexLogger.propertiesChanged : called at time : " + formatedTime);
+ String formattedTime = sdf.format(Date.from(startTime));
+ displayMessage("FlexLogger.propertiesChanged : called at time : " + formattedTime);
displayMessage("FlexLogger.propertiesChanged : debugLevel : " + debugLevel);
if (changedKeys != null) {
startTime = timeCheck("Second", startTime);
}
- if (errorCount != 0) {
+ if (errorCount > 0) {
String msg = " DB Audit: " + errorCount
+ " errors found. A large number of errors may indicate DB replication has stopped";
logger.error(MessageCodes.ERROR_AUDIT, msg);
/**
* class DbDao provides the inteface to the DBs for the purpose of audits.
- *
*/
public class DbDao {
private static final Logger logger = FlexLogger.getLogger();
private String nodeType;
private Properties properties = null;
- private EntityManagerFactory emf;
+ private final EntityManagerFactory emf;
/*
* Supports designation serialization.
* DB SELECT String.
*/
private static final String SELECT_STRING = "Select i from IntegrityAuditEntity i "
- + "where i.resourceName=:rn and i.persistenceUnit=:pu";
+ + "where i.resourceName=:rn and i.persistenceUnit=:pu";
/**
* DbDao Constructor.
*
- * @param resourceName the resource name
+ * @param resourceName the resource name
* @param persistenceUnit the persistence unit
- * @param properties the properties
+ * @param properties the properties
* @throws IntegrityAuditException if an error occurs
*/
public DbDao(String resourceName, String persistenceUnit, Properties properties) throws IntegrityAuditException {
/**
* DbDao Constructor.
*
- * @param resourceName the resource name
+ * @param resourceName the resource name
* @param persistenceUnit the persistence unit
- * @param properties the properties
- * @param altDbUrl may be {@code null}
+ * @param properties the properties
+ * @param altDbUrl may be {@code null}
* @throws IntegrityAuditException if an error occurs
*/
protected DbDao(String resourceName, String persistenceUnit, Properties properties, String altDbUrl)
- throws IntegrityAuditException {
+ throws IntegrityAuditException {
logger.debug("DbDao contructor: enter");
validateProperties(resourceName, persistenceUnit, properties);
/**
* validateProperties will validate the properties.
*
- * @param resourceName the rseource name
+ * @param resourceName the rseource name
* @param persistenceUnit the persistence unit
- * @param properties the properties
+ * @param properties the properties
* @throws IntegrityAuditPropertiesException if an error occurs
*/
private void validateProperties(String resourceName, String persistenceUnit, Properties properties)
- throws IntegrityAuditPropertiesException {
+ throws IntegrityAuditPropertiesException {
var badparams = new StringBuilder();
if (IntegrityAudit.parmsAreBad(resourceName, persistenceUnit, properties, badparams)) {
String msg = "DbDao: Bad parameters: badparams" + badparams;
this.nodeType = properties.getProperty(IntegrityAuditProperties.NODE_TYPE).trim().toLowerCase();
this.properties = properties;
logger.debug("DbDao.assignProperties: exit:" + "\nresourceName: " + this.resourceName + "\npersistenceUnit: "
- + this.persistenceUnit + "\nproperties: " + this.properties);
+ + this.persistenceUnit + "\nproperties: " + this.properties);
}
/**
HashMap<Object, Object> resultMap = new HashMap<>();
var em = emf.createEntityManager();
try {
- var cb = em.getCriteriaBuilder();
- CriteriaQuery<Object> cq = cb.createQuery();
- Root<?> rootEntry = cq.from(Class.forName(className));
- CriteriaQuery<Object> all = cq.select(rootEntry);
- TypedQuery<Object> allQuery = em.createQuery(all);
- List<Object> objectList = allQuery.getResultList();
- // Now create the map
-
- var util = emf.getPersistenceUnitUtil();
- for (Object o : objectList) {
- Object key = util.getIdentifier(o);
- resultMap.put(key, o);
- }
+ getObjectsFromCriteriaBuilder(className, emf, em, resultMap);
} catch (Exception e) {
logger.error("getAllEntries encountered exception: ", e);
}
* getAllMyEntries gets all entries for a class.
*
* @param className the name of the class
- * @param keySet the keys to get the entries for
+ * @param keySet the keys to get the entries for
* @return the map of requested entries
*/
public Map<Object, Object> getAllMyEntries(String className, Set<Object> keySet) {
}
/**
- * getAllEntries gets all entriesfor a particular persistence unit adn className.
+ * getAllEntries gets all entries for a particular persistence unit adn className.
*
* @param persistenceUnit the persistence unit
- * @param properties the properties
- * @param className the class name
+ * @param properties the properties
+ * @param className the class name
* @return the map of entries
*/
public Map<Object, Object> getAllEntries(String persistenceUnit, Properties properties, String className) {
var theEmf = Persistence.createEntityManagerFactory(persistenceUnit, properties);
var em = theEmf.createEntityManager();
+
try {
- var cb = em.getCriteriaBuilder();
- CriteriaQuery<Object> cq = cb.createQuery();
- Root<?> rootEntry = cq.from(Class.forName(className));
- CriteriaQuery<Object> all = cq.select(rootEntry);
- TypedQuery<Object> allQuery = em.createQuery(all);
- List<Object> objectList = allQuery.getResultList();
-
- var util = theEmf.getPersistenceUnitUtil();
- for (Object o : objectList) {
- Object key = util.getIdentifier(o);
- resultMap.put(key, o);
- }
+ getObjectsFromCriteriaBuilder(className, theEmf, em, resultMap);
} catch (Exception e) {
logger.error("getAllEntries encountered exception:", e);
}
* getAllEntries gets all entries for a persistence unit.
*
* @param persistenceUnit the persistence unit
- * @param properties the properties
- * @param className the class name
- * @param keySet the keys
+ * @param properties the properties
+ * @param className the class name
+ * @param keySet the keys
* @return the map of entries
*/
public Map<Object, Object> getAllEntries(String persistenceUnit, Properties properties, String className,
- Set<Object> keySet) {
+ Set<Object> keySet) {
logger.debug("getAllEntries: Entering, persistenceUnit=" + persistenceUnit + ",\n properties= " + properties
- + ",\n className=" + className + ",\n keySet= " + keySet);
+ + ",\n className=" + className + ",\n keySet= " + keySet);
var theEmf = Persistence.createEntityManagerFactory(persistenceUnit, properties);
var em = theEmf.createEntityManager();
HashMap<Object, Object> resultMap = new HashMap<>();
* unit and node type.
*
* @param persistenceUnit the persistence unit
- * @param nodeType the node type
+ * @param nodeType the node type
* @return the list of IntegrityAuditEntity
* @throws DbDaoTransactionException if an error occurs
*/
@SuppressWarnings("unchecked")
public List<IntegrityAuditEntity> getIntegrityAuditEntities(String persistenceUnit, String nodeType)
- throws DbDaoTransactionException {
+ throws DbDaoTransactionException {
logger.debug("getIntegrityAuditEntities: Entering, persistenceUnit=" + persistenceUnit + ",\n nodeType= "
- + nodeType);
+ + nodeType);
try {
- var em = emf.createEntityManager();
- // Start a transaction
- EntityTransaction et = em.getTransaction();
+ List<IntegrityAuditEntity> iaeList;
+ try (var em = emf.createEntityManager()) {
+ // Start a transaction
+ EntityTransaction et = em.getTransaction();
- et.begin();
+ et.begin();
- // if IntegrityAuditEntity entry exists for resourceName and PU, update it. If not
- // found, create a new entry
- var iaequery = em
+ // if IntegrityAuditEntity entry exists for resourceName and PU, update it. If not
+ // found, create a new entry
+ var iaequery = em
.createQuery("Select i from IntegrityAuditEntity i where i.persistenceUnit=:pu and i.nodeType=:nt");
- iaequery.setParameter("pu", persistenceUnit);
- iaequery.setParameter("nt", nodeType);
+ iaequery.setParameter("pu", persistenceUnit);
+ iaequery.setParameter("nt", nodeType);
- List<IntegrityAuditEntity> iaeList = iaequery.getResultList();
+ iaeList = iaequery.getResultList();
- // commit transaction
- et.commit();
- em.close();
+ // commit transaction
+ et.commit();
+ }
logger.debug("getIntegrityAuditEntities: Exit, iaeList=" + iaeList);
return iaeList;
} catch (Exception e) {
// refresh the object from DB in case cached data was returned
em.refresh(iae);
logger.info(RESOURCE_MESSAGE + this.resourceName + WITH_PERSISTENCE_MESSAGE + this.persistenceUnit
- + " exists");
+ + " exists");
} else {
// If it does not exist, log an error
logger.error("Attempting to setLastUpdated" + " on an entry that does not exist: resource "
- + this.resourceName + WITH_PERSISTENCE_MESSAGE + this.persistenceUnit);
+ + this.resourceName + WITH_PERSISTENCE_MESSAGE + this.persistenceUnit);
}
});
}
*/
public IntegrityAuditEntity getIntegrityAuditEntity(long id) throws DbDaoTransactionException {
try {
- var em = emf.createEntityManager();
+ IntegrityAuditEntity iae;
+ try (var em = emf.createEntityManager()) {
- // Start a transaction
- EntityTransaction et = em.getTransaction();
+ // Start a transaction
+ EntityTransaction et = em.getTransaction();
- et.begin();
+ et.begin();
- IntegrityAuditEntity iae = em.find(IntegrityAuditEntity.class, id);
+ iae = em.find(IntegrityAuditEntity.class, id);
- et.commit();
- em.close();
+ et.commit();
+ }
return iae;
} catch (Exception e) {
* Register the IntegrityAudit instance.
*
* @param altDbUrl alternate DB URL to be placed into the record, or {@code null} to use the
- * default
+ * default
*/
private void register(String altDbUrl) throws DbDaoTransactionException {
// refresh the object from DB in case cached data was returned
em.refresh(iae2);
logger.info(RESOURCE_MESSAGE + this.resourceName + WITH_PERSISTENCE_MESSAGE + this.persistenceUnit
- + " exists and entry be updated");
+ + " exists and entry be updated");
} else {
// If it does not exist, we also must add teh resourceName, persistenceUnit and
// designated values
logger.info("Adding resource " + resourceName + WITH_PERSISTENCE_MESSAGE + this.persistenceUnit
- + " to IntegrityAuditEntity table");
+ + " to IntegrityAuditEntity table");
iae2 = new IntegrityAuditEntity();
iae2.setResourceName(this.resourceName);
iae2.setPersistenceUnit(this.persistenceUnit);
/**
* Set designated.
*
- * @param resourceName the resource name
+ * @param resourceName the resource name
* @param persistenceUnit the persistence unit
- * @param desig true if is designated
+ * @param desig true if is designated
* @throws DbDaoTransactionException if an error occurs
*/
public void setDesignated(String resourceName, String persistenceUnit, boolean desig)
- throws DbDaoTransactionException {
+ throws DbDaoTransactionException {
logger.debug("setDesignated: enter, resourceName: " + resourceName + ", persistenceUnit: " + persistenceUnit
- + ", designated: " + desig);
+ + ", designated: " + desig);
updateIae("setDesignated", resourceName, persistenceUnit, (em, iae) -> {
// refresh the object from DB in case cached data was returned
em.refresh(iae);
logger.info(RESOURCE_MESSAGE + resourceName + WITH_PERSISTENCE_MESSAGE + persistenceUnit
- + " exists and designated be updated");
+ + " exists and designated be updated");
iae.setDesignated(desig);
em.persist(iae);
} else {
// If it does not exist, log an error
logger.error("Attempting to setDesignated(" + desig + ") on an entry that does not exist:"
- + " resource " + resourceName + WITH_PERSISTENCE_MESSAGE + persistenceUnit);
+ + " resource " + resourceName + WITH_PERSISTENCE_MESSAGE + persistenceUnit);
}
});
/**
* Queries for an audit entity and then updates it using an "updater" function.
*
- * @param methodName name of the method that invoked this
- * @param resourceName the resource name
+ * @param methodName name of the method that invoked this
+ * @param resourceName the resource name
* @param persistenceUnit the persistence unit
- * @param updater function to update the entity; the argument will be the entity to be
- * updated, or {@code null} if the entity is not found
+ * @param updater function to update the entity; the argument will be the entity to be
+ * updated, or {@code null} if the entity is not found
* @return the entity that was found, or {@code null} if the entity is not found
* @throws DbDaoTransactionException if an error occurs
*/
private IntegrityAuditEntity updateIae(String methodName, String resourceName, String persistenceUnit,
- BiConsumer<EntityManager, IntegrityAuditEntity> updater) throws DbDaoTransactionException {
+ BiConsumer<EntityManager, IntegrityAuditEntity> updater)
+ throws DbDaoTransactionException {
try {
- var em = emf.createEntityManager();
+ IntegrityAuditEntity iae;
+ try (var em = emf.createEntityManager()) {
- // Start a transaction
- EntityTransaction et = em.getTransaction();
+ // Start a transaction
+ EntityTransaction et = em.getTransaction();
- et.begin();
+ et.begin();
- // if IntegrityAuditEntity entry exists for resourceName and PU, update it. If not
- // found, create a new entry
- TypedQuery<IntegrityAuditEntity> iaequery = em.createQuery(SELECT_STRING, IntegrityAuditEntity.class);
- iaequery.setParameter("rn", resourceName);
- iaequery.setParameter("pu", persistenceUnit);
+ // if IntegrityAuditEntity entry exists for resourceName and PU, update it. If not
+ // found, create a new entry
+ TypedQuery<IntegrityAuditEntity> iaequery = em.createQuery(SELECT_STRING, IntegrityAuditEntity.class);
+ iaequery.setParameter("rn", resourceName);
+ iaequery.setParameter("pu", persistenceUnit);
- List<IntegrityAuditEntity> iaeList = iaequery.getResultList();
- IntegrityAuditEntity iae;
+ List<IntegrityAuditEntity> iaeList = iaequery.getResultList();
- if (!iaeList.isEmpty()) {
- // ignores multiple results
- iae = iaeList.get(0);
+ if (!iaeList.isEmpty()) {
+ // ignores multiple results
+ iae = iaeList.get(0);
- } else {
- // If it does not exist
- iae = null;
- }
+ } else {
+ // If it does not exist
+ iae = null;
+ }
- updater.accept(em, iae);
+ updater.accept(em, iae);
- // close the transaction
- et.commit();
- // close the EntityManager
- em.close();
+ // close the transaction
+ et.commit();
+ }
return iae;
*/
public void setLastUpdated() throws DbDaoTransactionException {
logger.debug("setLastUpdated: enter, resourceName: " + this.resourceName + ", persistenceUnit: "
- + this.persistenceUnit);
+ + this.persistenceUnit);
updateIae("setLastUpdated", this.resourceName, this.persistenceUnit, (em, iae) -> {
// refresh the object from DB in case cached data was returned
em.refresh(iae);
logger.info(RESOURCE_MESSAGE + this.resourceName + WITH_PERSISTENCE_MESSAGE + this.persistenceUnit
- + " exists and lastUpdated be updated");
+ + " exists and lastUpdated be updated");
iae.setLastUpdated(AuditorTime.getInstance().getDate());
em.persist(iae);
} else {
// If it does not exist, log an error
logger.error("Attempting to setLastUpdated" + " on an entry that does not exist:" + " resource "
- + this.resourceName + WITH_PERSISTENCE_MESSAGE + this.persistenceUnit);
+ + this.resourceName + WITH_PERSISTENCE_MESSAGE + this.persistenceUnit);
}
});
}
if (!IntegrityAudit.isUnitTesting()) {
String msg = DBDAO_MESSAGE + "deleteAllIntegrityAuditEntities() "
- + "should only be invoked during JUnit testing";
+ + "should only be invoked during JUnit testing";
logger.error(msg);
throw new DbDaoTransactionException(msg);
}
- var em = emf.createEntityManager();
- // Start a transaction
- EntityTransaction et = em.getTransaction();
+ int returnCode;
+ try (var em = emf.createEntityManager()) {
+ // Start a transaction
+ EntityTransaction et = em.getTransaction();
- et.begin();
+ et.begin();
- // if IntegrityAuditEntity entry exists for resourceName and PU, update it. If not
- // found, create a new entry
- var iaequery = em.createQuery("Delete from IntegrityAuditEntity");
+ // if IntegrityAuditEntity entry exists for resourceName and PU, update it. If not
+ // found, create a new entry
+ var iaequery = em.createQuery("Delete from IntegrityAuditEntity");
- int returnCode = iaequery.executeUpdate();
+ returnCode = iaequery.executeUpdate();
- // commit transaction
- et.commit();
- em.close();
+ // commit transaction
+ et.commit();
+ }
logger.info("deleteAllIntegrityAuditEntities: returnCode=" + returnCode);
* designations from interleaved changeDesignated() invocations from different resources
* (entities), because it prevents "dirty" and "non-repeatable" reads.
*
- * <p>See http://www.objectdb.com/api/java/jpa/LockModeType
+ * <p>See www.objectdb.com/api/java/jpa/LockModeType
*
* <p>and
*
- * <p>http://stackoverflow.com/questions/2120248/how-to-synchronize-a-static-
+ * <p>stackoverflow.com/questions/2120248/how-to-synchronize-a-static-
* variable-among-threads-running-different-instances-o
*/
public void changeDesignated(String resourceName, String persistenceUnit, String nodeType)
- throws DbDaoTransactionException {
+ throws DbDaoTransactionException {
if (logger.isDebugEnabled()) {
logger.debug("changeDesignated: Entering, resourceName=" + resourceName + ", persistenceUnit="
- + persistenceUnit + ", nodeType=" + nodeType);
+ + persistenceUnit + ", nodeType=" + nodeType);
}
long startTime = AuditorTime.getInstance().getMillis();
synchronized (lock) {
-
- EntityManager em = null;
- try {
-
- em = emf.createEntityManager();
- em.getTransaction().begin();
-
- /*
- * Define query
- */
- var query = em.createQuery(
+ try (var em = emf.createEntityManager()) {
+ try {
+ em.getTransaction().begin();
+
+ /*
+ * Define query
+ */
+ var query = em.createQuery(
"Select i from IntegrityAuditEntity i where i.persistenceUnit=:pu and i.nodeType=:nt");
- query.setParameter("pu", persistenceUnit);
- query.setParameter("nt", nodeType);
-
- /*
- * Execute query using pessimistic write lock. This ensures that if anyone else is
- * currently reading the records we'll throw a LockTimeoutException.
- */
- setDesignatedEntity(resourceName, query);
-
- if (logger.isDebugEnabled()) {
- logger.debug("changeDesignated: Committing designation to resourceName=" + resourceName);
- }
- em.getTransaction().commit();
-
- /*
- * If we get a LockTimeoutException, no harm done really. We'll probably be
- * successful on the next attempt. The odds of another DbDao instance on this entity
- * or another entity attempting a simultaneous IntegrityAuditEntity table
- * read/update are pretty slim (we're only in this method for two or three
- * milliseconds)
- */
- } catch (Exception e) {
- if (em != null) {
- em.getTransaction().rollback();
-
- String msg = "DbDao: changeDesignated() caught Exception, message=" + e.getMessage();
- logger.error(msg + e);
- throw new DbDaoTransactionException(msg, e);
- } else {
- String msg = "DbDao: changeDesignated() caught Exception, message="
+ query.setParameter("pu", persistenceUnit);
+ query.setParameter("nt", nodeType);
+
+ /*
+ * Execute query using pessimistic write lock. This ensures that if anyone else is
+ * currently reading the records we'll throw a LockTimeoutException.
+ */
+ setDesignatedEntity(resourceName, query);
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("changeDesignated: Committing designation to resourceName=" + resourceName);
+ }
+ em.getTransaction().commit();
+
+ /*
+ * If we get a LockTimeoutException, no harm done really. We'll probably be
+ * successful on the next attempt. The odds of another DbDao instance on this entity
+ * or another entity attempting a simultaneous IntegrityAuditEntity table
+ * read/update are pretty slim (we're only in this method for two or three
+ * milliseconds)
+ */
+ } catch (Exception e) {
+ String errorMsg;
+ try {
+ em.getTransaction().rollback();
+ errorMsg = "DbDao: changeDesignated() caught Exception, message=" + e.getMessage();
+ } catch (Exception rollbackException) {
+ errorMsg = "DbDao: changeDesignated() caught Exception, message="
+ e.getMessage() + ". Error rolling back transaction.";
- logger.error(msg + e);
- throw new DbDaoTransactionException(msg, e);
+ }
+ logger.error(errorMsg + e);
+ throw new DbDaoTransactionException(errorMsg, e);
}
}
if (logger.isDebugEnabled()) {
logger.debug("changeDesignated: Exiting; time expended="
- + (AuditorTime.getInstance().getMillis() - startTime) + "ms");
+ + (AuditorTime.getInstance().getMillis() - startTime) + "ms");
}
}
private void setDesignatedEntity(String resourceName, Query query) {
for (Object o : query.getResultList()) {
- if (!(o instanceof IntegrityAuditEntity)) {
+ if (!(o instanceof IntegrityAuditEntity integrityAuditEntity)) {
continue;
}
- var integrityAuditEntity = (IntegrityAuditEntity) o;
if (integrityAuditEntity.getResourceName().equals(resourceName)) {
if (logger.isDebugEnabled()) {
logger.debug("changeDesignated: Designating resourceName="
- + integrityAuditEntity.getResourceName());
+ + integrityAuditEntity.getResourceName());
}
integrityAuditEntity.setDesignated(true);
} else {
if (logger.isDebugEnabled()) {
logger.debug("changeDesignated: Removing designation from resourceName="
- + integrityAuditEntity.getResourceName());
+ + integrityAuditEntity.getResourceName());
}
integrityAuditEntity.setDesignated(false);
}
}
}
+ /**
+ * Collects all objects from a criteria builder based on className.
+ *
+ * @param className type of objects for resultMap
+ * @param emf the entity manager factory to be used
+ * @param em entity manager to be used
+ * @param resultMap the result map for objects queried
+ * @throws ClassNotFoundException if class for criteria builder doesn't exist
+ */
+ private void getObjectsFromCriteriaBuilder(String className, EntityManagerFactory emf, EntityManager em,
+ HashMap<Object, Object> resultMap)
+ throws ClassNotFoundException {
+ var cb = em.getCriteriaBuilder();
+ CriteriaQuery<Object> cq = cb.createQuery();
+ Root<?> rootEntry = cq.from(Class.forName(className));
+ CriteriaQuery<Object> all = cq.select(rootEntry);
+ TypedQuery<Object> allQuery = em.createQuery(all);
+ List<Object> objectList = allQuery.getResultList();
+ // Now create the map
+
+ var util = emf.getPersistenceUnitUtil();
+ for (Object o : objectList) {
+ Object key = util.getIdentifier(o);
+ resultMap.put(key, o);
+ }
+ }
+
}
import jakarta.persistence.GeneratedValue;
import jakarta.persistence.GenerationType;
import jakarta.persistence.Id;
-import jakarta.persistence.NamedQueries;
import jakarta.persistence.NamedQuery;
import jakarta.persistence.PrePersist;
import jakarta.persistence.PreUpdate;
import jakarta.persistence.TemporalType;
import java.io.IOException;
import java.io.ObjectOutputStream;
+import java.io.Serial;
import java.io.Serializable;
import java.util.Date;
import lombok.AccessLevel;
@Entity
@Table(name = "IntegrityAuditEntity")
-@NamedQueries({
- @NamedQuery(name = " IntegrityAuditEntity.findAll", query = "SELECT e FROM IntegrityAuditEntity e "),
- @NamedQuery(name = "IntegrityAuditEntity.deleteAll", query = "DELETE FROM IntegrityAuditEntity WHERE 1=1")
-})
+@NamedQuery(name = " IntegrityAuditEntity.findAll", query = "SELECT e FROM IntegrityAuditEntity e ")
+@NamedQuery(name = "IntegrityAuditEntity.deleteAll", query = "DELETE FROM IntegrityAuditEntity WHERE 1=1")
@NoArgsConstructor
@Getter
@Setter
public class IntegrityAuditEntity implements Serializable {
+
+ @Serial
private static final long serialVersionUID = 1L;
@Getter
private static final String PROPERTY_EXCEPTION_STRING = "IntegrityMonitor Property Exception: ";
private static final String EXCEPTION_STRING = "IntegrityMonitor threw exception.";
private static final String STATE_CHECK_STRING =
- "IntegrityMonitor.stateCheck(): " + "Failed to disableFail dependent resource = ";
+ "IntegrityMonitor.stateCheck(): " + "Failed to disableFail dependent resource = ";
private static final String RESOURCE_STRING = "Resource ";
private static final String LC_RESOURCE_STRING = "resource";
private String dependencyCheckErrorMsg = "";
// The entity manager factory for JPA access
- private EntityManagerFactory emf;
- private EntityManager em;
+ private final EntityManagerFactory emf;
+ private final EntityManager em;
// Persistence Unit for JPA
public static final String PERSISTENCE_UNIT = "operationalPU";
private static long writeFpcIntervalMs = toMillis(IntegrityMonitorProperties.DEFAULT_WRITE_FPC_INTERVAL);
// check the health of dependencies
private static long checkDependencyIntervalMs =
- toMillis(IntegrityMonitorProperties.DEFAULT_CHECK_DEPENDENCY_INTERVAL);
+ toMillis(IntegrityMonitorProperties.DEFAULT_CHECK_DEPENDENCY_INTERVAL);
// A lead subsystem will have dependency groups with resource names in the
// properties file.
/**
* IntegrityMonitor constructor. It is invoked from the getInstance() method in this class or from the constructor
- * of a child or sub-class. A class can extend the IntegrityMonitor class if there is a need to override any of the
+ * of a child or subclass. A class can extend the IntegrityMonitor class if there is a need to override any of the
* base methods (ex. subsystemTest()). Only one instance is allowed to be created per resource name.
*
* @param resourceName The resource name of the resource
- * @param properties a set of properties passed in from the resource
+ * @param properties a set of properties passed in from the resource
* @throws IntegrityMonitorException if any errors are encountered in the constructor
*/
protected IntegrityMonitor(String resourceName, Properties properties) throws IntegrityMonitorException {
// singleton check since this constructor can be called from a child or
- // sub-class
+ // subclass
if (instance != null) {
var msg = "IM object exists and only one instance allowed";
logger.error("{}", msg);
fquery.setParameter("rn", resourceName);
List<ForwardProgressEntity> fpList =
- fquery.setLockMode(LockModeType.NONE).setFlushMode(FlushModeType.COMMIT).getResultList();
- ForwardProgressEntity fpx = null;
+ fquery.setLockMode(LockModeType.NONE).setFlushMode(FlushModeType.COMMIT).getResultList();
+ ForwardProgressEntity fpx;
if (!fpList.isEmpty()) {
// ignores multiple results
fpx = fpList.get(0);
em.refresh(fpx);
if (logger.isDebugEnabled()) {
logger.debug("Resource {} exists and will be updated - old fpc= {}, lastUpdated= {}", resourceName,
- fpx.getFpcCount(), fpx.getLastUpdated());
+ fpx.getFpcCount(), fpx.getLastUpdated());
}
fpx.setFpcCount(fpCounter);
} else {
protected void createOrUpdateResourceReg(String resourceName, String jmxUrl, EntityTransaction et) {
TypedQuery<ResourceRegistrationEntity> rquery =
- em.createQuery("Select r from ResourceRegistrationEntity r where r.resourceName=:rn",
- ResourceRegistrationEntity.class);
+ em.createQuery("Select r from ResourceRegistrationEntity r where r.resourceName=:rn",
+ ResourceRegistrationEntity.class);
rquery.setParameter("rn", resourceName);
List<ResourceRegistrationEntity> rrList =
- rquery.setLockMode(LockModeType.NONE).setFlushMode(FlushModeType.COMMIT).getResultList();
- ResourceRegistrationEntity rrx = null;
+ rquery.setLockMode(LockModeType.NONE).setFlushMode(FlushModeType.COMMIT).getResultList();
+ ResourceRegistrationEntity rrx;
if (!rrList.isEmpty()) {
// ignores multiple results
rrx = rrList.get(0);
em.refresh(rrx);
if (logger.isDebugEnabled()) {
logger.debug("Resource {} exists and will be updated - old url= {}, createdDate={}", resourceName,
- rrx.getResourceUrl(), rrx.getCreatedDate());
+ rrx.getResourceUrl(), rrx.getCreatedDate());
}
rrx.setLastUpdated(MonitorTime.getInstance().getDate());
} else {
try {
new ComponentAdmin(resourceName, this, stateManager);
} catch (Exception e) {
- logger.error("ComponentAdmin constructor exception: {}", e.toString(), e);
+ logger.error("ComponentAdmin constructor exception: {}", e, e);
}
}
* instance is allowed to be created per resource name.
*
* @param resourceName The resource name of the resource
- * @param properties a set of properties passed in from the resource
+ * @param properties a set of properties passed in from the resource
* @return The new instance of IntegrityMonitor
* @throws IntegrityMonitorException if unable to create jmx url or the constructor returns an exception
*/
public static IntegrityMonitor getInstance(String resourceName, Properties properties)
- throws IntegrityMonitorException {
+ throws IntegrityMonitorException {
synchronized (getInstanceLock) {
logger.debug("getInstance() called - resourceName= {}", resourceName);
logger.debug("getInstance() called");
if (instance == null) {
String msg = "No IntegrityMonitor instance exists."
- + " Please use the method IntegrityMonitor.getInstance(String resourceName, Properties properties)";
+ + " Please use the method IntegrityMonitor.getInstance(String resourceName, Properties properties)";
throw new IntegrityMonitorPropertiesException(msg);
} else {
return instance;
if (fpm.isAlive()) {
logger.error("IntegrityMonitor.deleteInstance() Failed to kill FPManager thread");
throw new IntegrityMonitorException(
- "IntegrityMonitor.deleteInstance() Failed to kill FPManager thread");
+ "IntegrityMonitor.deleteInstance() Failed to kill FPManager thread");
}
instance = null;
}
// check standby state and throw exception if cold standby
if ((stateManager.getStandbyStatus() != null)
- && stateManager.getStandbyStatus().equals(StateManagement.COLD_STANDBY)) {
+ && stateManager.getStandbyStatus().equals(StateManagement.COLD_STANDBY)) {
String msg = RESOURCE_STRING + resourceName + " is cold standby";
logger.debug("{}", msg);
throw new StandbyStatusException("IntegrityMonitor Standby Status Exception: " + msg);
String errorMsg =
withinTransaction(dep + ": ForwardProgressEntity DB operation failed with exception: ", () -> {
TypedQuery<ForwardProgressEntity> query = em.createQuery(
- "Select p from ForwardProgressEntity p where p.resourceName=:resource",
- ForwardProgressEntity.class);
+ "Select p from ForwardProgressEntity p where p.resourceName=:resource",
+ ForwardProgressEntity.class);
query.setParameter(LC_RESOURCE_STRING, dep);
List<ForwardProgressEntity> fpList = query.setLockMode(LockModeType.NONE)
- .setFlushMode(FlushModeType.COMMIT).getResultList();
+ .setFlushMode(FlushModeType.COMMIT).getResultList();
if (!fpList.isEmpty()) {
// exists
// returned
em.refresh(forwardProgressEntity.get());
logger.debug("Found entry in ForwardProgressEntity table for dependent Resource={}",
- dep);
+ dep);
return null;
} else {
}
});
- if (errorMsg == null) {
+ if (StringUtils.isEmpty(errorMsg)) {
errorMsg = withinTransaction(dep + ": StateManagementEntity DB read failed with exception: ", () -> {
// query if StateManagement entry exists for dependent resource
TypedQuery<StateManagementEntity> query =
- em.createQuery("Select p from StateManagementEntity p where p.resourceName=:resource",
- StateManagementEntity.class);
+ em.createQuery("Select p from StateManagementEntity p where p.resourceName=:resource",
+ StateManagementEntity.class);
query.setParameter(LC_RESOURCE_STRING, dep);
List<StateManagementEntity> smList =
- query.setLockMode(LockModeType.NONE).setFlushMode(FlushModeType.COMMIT).getResultList();
+ query.setLockMode(LockModeType.NONE).setFlushMode(FlushModeType.COMMIT).getResultList();
if (!smList.isEmpty()) {
// exist
stateManagementEntity.set(smList.get(0));
}
// verify that the ForwardProgress is current (check last_updated)
- if (errorMsg == null) {
+ if (StringUtils.isEmpty(errorMsg)) {
checkForwardProgress(dep, forwardProgressEntity.get(), stateManagementEntity.get());
}
// check operation, admin and standby states of dependent resource
- if (errorMsg == null) {
+ if (StringUtils.isEmpty(errorMsg)) {
errorMsg = checkDependentStates(dep, stateManagementEntity.get());
}
/**
* Runs an action within a transaction.
*
- * @param exMsg message to log and return if an exception occurs
+ * @param exMsg message to log and return if an exception occurs
* @param action action to apply; returns non-null if an error occurs
* @return {@code null} if success, or an error message otherwise
*/
private String withinTransaction(String exMsg, Supplier<String> action) {
- String errorMsg = null;
-
// Start a transaction
EntityTransaction et = em.getTransaction();
et.begin();
try {
- errorMsg = action.get();
- if (errorMsg != null) {
- logger.error("{}", errorMsg);
- }
-
+ var errorMsg = action.get();
+ logger.error("{}", errorMsg);
synchronized (imFlushLock) {
et.commit();
}
-
+ return errorMsg;
} catch (RuntimeException ex) {
// log an error
- errorMsg = exMsg;
- logger.error("{}", errorMsg, ex);
+ logger.error("{}", exMsg, ex);
synchronized (imFlushLock) {
if (et.isActive()) {
et.rollback();
}
}
+ return exMsg;
}
-
- return errorMsg;
}
private void checkForwardProgress(String dep, ForwardProgressEntity forwardProgressEntity,
- StateManagementEntity stateManagementEntity) {
+ StateManagementEntity stateManagementEntity) {
if (forwardProgressEntity != null && stateManagementEntity != null) {
var date = MonitorTime.getInstance().getDate();
long diffMs = date.getTime() - forwardProgressEntity.getLastUpdated().getTime();
}
} else {
+ String msg;
if (forwardProgressEntity == null) {
- String msg = STATE_CHECK_STRING + dep + "; " + " forwardProgressEntity == null.";
- logger.error("{}", msg);
+ msg = STATE_CHECK_STRING + dep + "; " + " forwardProgressEntity == null.";
} else {
- String msg = STATE_CHECK_STRING + dep + "; " + " stateManagementEntity == null.";
- logger.error("{}", msg);
+ msg = STATE_CHECK_STRING + dep + "; " + " stateManagementEntity == null.";
}
+ logger.error("{}", msg);
}
}
if (stateManagementEntity != null) {
if ((stateManager.getAdminState() != null)
- && stateManagementEntity.getAdminState().equals(StateManagement.LOCKED)) {
+ && stateManagementEntity.getAdminState().equals(StateManagement.LOCKED)) {
errorMsg = dep + ": resource is administratively locked";
logger.error("{}", errorMsg);
} else if ((stateManager.getOpState() != null)
- && stateManagementEntity.getOpState().equals(StateManagement.DISABLED)) {
+ && stateManagementEntity.getOpState().equals(StateManagement.DISABLED)) {
errorMsg = dep + ": resource is operationally disabled";
logger.error("{}", errorMsg);
} else if ((stateManager.getStandbyStatus() != null)
- && stateManagementEntity.getStandbyStatus().equals(StateManagement.COLD_STANDBY)) {
+ && stateManagementEntity.getStandbyStatus().equals(StateManagement.COLD_STANDBY)) {
errorMsg = dep + ": resource is cold standby";
logger.error("{}", errorMsg);
}
fquery.setParameter("rn", dep);
List<ForwardProgressEntity> fpList =
- fquery.setLockMode(LockModeType.NONE).setFlushMode(FlushModeType.COMMIT).getResultList();
+ fquery.setLockMode(LockModeType.NONE).setFlushMode(FlushModeType.COMMIT).getResultList();
ForwardProgressEntity fpx;
if (!fpList.isEmpty()) {
// ignores multiple results
em.refresh(fpx);
if (logger.isDebugEnabled()) {
logger.debug("Dependent resource {} - fpc= {}, lastUpdated={}", dep, fpx.getFpcCount(),
- fpx.getLastUpdated());
+ fpx.getLastUpdated());
}
long currTime = MonitorTime.getInstance().getMillis();
// if dependent resource FPC has not been updated, consider it
withinTransaction("getAllForwardProgessEntity DB read failed with exception: ", () -> {
var fquery = em.createQuery("Select e from ForwardProgressEntity e");
fquery.setLockMode(LockModeType.NONE).setFlushMode(FlushModeType.COMMIT).getResultList()
- .forEach(obj -> fpList.add((ForwardProgressEntity) obj));
+ .forEach(obj -> fpList.add((ForwardProgressEntity) obj));
return null;
});
var index = 0;
for (ForwardProgressEntity fpe : fpList) {
logger.debug("getAllForwardProgressEntity: fpList.get({}).getResourceName(): {}", index++,
- fpe.getResourceName());
+ fpe.getResourceName());
}
return fpList;
private String getJmxUrlFromDb(String dep, AtomicReference<String> jmxUrl) {
// query if ResourceRegistration entry exists for resourceName
TypedQuery<ResourceRegistrationEntity> rquery =
- em.createQuery("Select r from ResourceRegistrationEntity r where r.resourceName=:rn",
- ResourceRegistrationEntity.class);
+ em.createQuery("Select r from ResourceRegistrationEntity r where r.resourceName=:rn",
+ ResourceRegistrationEntity.class);
rquery.setParameter("rn", dep);
List<ResourceRegistrationEntity> rrList =
- rquery.setLockMode(LockModeType.NONE).setFlushMode(FlushModeType.COMMIT).getResultList();
- ResourceRegistrationEntity rrx = null;
+ rquery.setLockMode(LockModeType.NONE).setFlushMode(FlushModeType.COMMIT).getResultList();
+ ResourceRegistrationEntity rrx;
if (!rrList.isEmpty()) {
// ignores multiple results
jmxAgentConnection = new JmxAgentConnection(jmxUrl);
MBeanServerConnection mbeanServer = jmxAgentConnection.getMBeanConnection();
ComponentAdminMBean admin =
- JMX.newMXBeanProxy(mbeanServer, ComponentAdmin.getObjectName(dep), ComponentAdminMBean.class);
+ JMX.newMXBeanProxy(mbeanServer, ComponentAdmin.getObjectName(dep), ComponentAdminMBean.class);
// invoke the test method via the jmx proxy
admin.test();
try {
if (logger.isDebugEnabled()) {
logger.debug("{}: There has been a subsystemTest failure with error:{} Updating this resource's "
- + "state to disableDependency", resourceName, e.getMessage());
+ + "state to disableDependency", resourceName, e.getMessage());
}
// Capture the subsystemTest failure info
appendSeparator(errorMsg);
/**
* Checks if a dependency group has an error.
*
- * @param group group to be checked
+ * @param group group to be checked
* @param errorMsg error messages are appended here
* @return {@code true} if the group has an error, {@code false} otherwise
*/
* Disables the dependency group.
*
* @param errorMsg error messages are appended to this
- * @param group group of interest
+ * @param group group of interest
* @return {@code true} if it was successfully disabled, {@code false} otherwise
*/
private boolean disableDependency(StringBuilder errorMsg, String group) {
try {
logger.debug("All dependents in group {} have failed their health check. Updating this "
- + "resource's state to disableDependency", group);
+ + "resource's state to disableDependency", group);
if (stateManager.getAvailStatus() == null
- || !((stateManager.getAvailStatus()).equals(StateManagement.DEPENDENCY)
- || (stateManager.getAvailStatus()).equals(StateManagement.DEPENDENCY_FAILED))) {
+ || !((stateManager.getAvailStatus()).equals(StateManagement.DEPENDENCY)
+ || (stateManager.getAvailStatus()).equals(StateManagement.DEPENDENCY_FAILED))) {
// Note: redundant calls are made by
// refreshStateAudit
this.stateManager.disableDependency();
private void dependenciesGood(StringBuilder errorMsg) {
try {
logger.debug("All dependency groups have at least one viable member. Updating this resource's state"
- + " to enableNoDependency");
+ + " to enableNoDependency");
if (stateManager.getAvailStatus() != null
- && ((stateManager.getAvailStatus()).equals(StateManagement.DEPENDENCY)
- || (stateManager.getAvailStatus()).equals(StateManagement.DEPENDENCY_FAILED))) {
+ && ((stateManager.getAvailStatus()).equals(StateManagement.DEPENDENCY)
+ || (stateManager.getAvailStatus()).equals(StateManagement.DEPENDENCY_FAILED))) {
// Note: redundant calls are made by
// refreshStateAudit
this.stateManager.enableNoDependency();
try {
logger.debug("There are no dependents. Updating this resource's state to enableNoDependency");
if (stateManager.getAvailStatus() != null
- && ((stateManager.getAvailStatus()).equals(StateManagement.DEPENDENCY)
- || (stateManager.getAvailStatus()).equals(StateManagement.DEPENDENCY_FAILED))) {
+ && ((stateManager.getAvailStatus()).equals(StateManagement.DEPENDENCY)
+ || (stateManager.getAvailStatus()).equals(StateManagement.DEPENDENCY_FAILED))) {
// Note: redundant calls are made by refreshStateAudit
this.stateManager.enableNoDependency();
}
}
private void appendSeparator(StringBuilder errorMsg) {
- if (errorMsg.length() != 0) {
+ if (!errorMsg.isEmpty()) {
errorMsg.append(',');
}
}
* Additional testing for subsystems that do not have a /test interface (for ex. 3rd party processes like elk). This
* method would be overridden by the subsystem.
*
- * @throws IntegrityMonitorException if an error occurs
*/
- public void subsystemTest() throws IntegrityMonitorException {
+ public void subsystemTest() {
// Testing provided by subsystem
logger.debug("IntegrityMonitor subsystemTest() OK");
}
// check standby state and throw exception if locked
if ((stateManager.getStandbyStatus() != null)
- && (stateManager.getStandbyStatus().equals(StateManagement.HOT_STANDBY)
- || stateManager.getStandbyStatus().equals(StateManagement.COLD_STANDBY))) {
+ && (stateManager.getStandbyStatus().equals(StateManagement.HOT_STANDBY)
+ || stateManager.getStandbyStatus().equals(StateManagement.COLD_STANDBY))) {
String msg = RESOURCE_STRING + resourceName + " is standby";
throw new StandbyStatusException("IntegrityMonitor Standby Status Exception: " + msg);
if (getAllNotWellMap() != null) {
if (!(getAllNotWellMap().isEmpty())) {
/*
- * An entity has reported that it is not well. We must not allow the the forward progress counter to
+ * An entity has reported that it is not well. We must not allow the forward progress counter to
* advance.
*/
var msg = new StringBuilder("allNotWellMap:");
buildMapString(msg, allNotWellMap);
- logger.error("endTransaction: allNotWellMap is NOT EMPTY. Not advancing forward"
- + "progress counter. \n{}\n", msg);
+ logger.error(
+ """
+ endTransaction: allNotWellMap is NOT EMPTY. Not advancing forward progress counter.
+ {}
+ """, msg);
return;
}
if (logger.isDebugEnabled() && getAllSeemsWellMap() != null && !(getAllSeemsWellMap().isEmpty())) {
var msg = new StringBuilder("allSeemsWellMap:");
buildMapString(msg, allSeemsWellMap);
- logger.debug("endTransaction: allNotWellMap IS EMPTY and allSeemsWellMap is NOT EMPTY. "
- + "Advancing forward progress counter. \n{}\n", msg);
+ logger.debug(
+ """
+ endTransaction: allNotWellMap IS EMPTY and allSeemsWellMap is NOT EMPTY.
+ Advancing forward progress counter.
+ {}
+ """, msg);
}
}
// increment local FPC
fquery.setParameter("rn", resourceName);
List<ForwardProgressEntity> fpList =
- fquery.setLockMode(LockModeType.NONE).setFlushMode(FlushModeType.COMMIT).getResultList();
+ fquery.setLockMode(LockModeType.NONE).setFlushMode(FlushModeType.COMMIT).getResultList();
ForwardProgressEntity fpx;
if (!fpList.isEmpty()) {
// ignores multiple results
em.refresh(fpx);
if (logger.isDebugEnabled()) {
logger.debug("Updating FP entry: Resource={}, fpcCount={}, lastUpdated={}, new fpcCount={}",
- resourceName, fpx.getFpcCount(), fpx.getLastUpdated(), fpCounter);
+ resourceName, fpx.getFpcCount(), fpx.getLastUpdated(), fpCounter);
}
fpx.setFpcCount(fpCounter);
em.persist(fpx);
setLong(prop, IntegrityMonitorProperties.FP_MONITOR_INTERVAL, value -> monitorIntervalMs = toMillis(value));
- setInt(prop, IntegrityMonitorProperties.FAILED_COUNTER_THRESHOLD, value -> failedCounterThreshold = value);
+ setInt(prop, value -> failedCounterThreshold = value);
setLong(prop, IntegrityMonitorProperties.TEST_TRANS_INTERVAL, value -> testTransIntervalMs = toMillis(value));
setLong(prop, IntegrityMonitorProperties.CHECK_DEPENDENCY_INTERVAL,
value -> checkDependencyIntervalMs = toMillis(value));
- // dependency_groups are a semi-colon separated list of groups
+ // dependency_groups are a semicolon separated list of groups
// each group is a comma separated list of resource names
// For ex. dependency_groups = site_1.pap_1,site_1.pap_2 ; site_1.pdp_1,
// site_1.pdp_2
value -> refreshStateAuditIntervalMs = value);
logger.debug("IntegrityMonitor.validateProperties(): Property values \nmaxFpcUpdateIntervalMs = {}\n",
- maxFpcUpdateIntervalMs);
+ maxFpcUpdateIntervalMs);
}
private static void setBoolean(Properties props, String propName, Consumer<Boolean> setter) {
return propValue.trim();
}
- private static void setInt(Properties props, String propName, IntConsumer setter) {
- String propValue = props.getProperty(propName);
+ private static void setInt(Properties props, IntConsumer setter) {
+ String propValue = props.getProperty(IntegrityMonitorProperties.FAILED_COUNTER_THRESHOLD);
if (StringUtils.isBlank(propValue)) {
return;
}
try {
setter.accept(Integer.parseInt(propValue.trim()));
} catch (NumberFormatException e) {
- logger.warn(IGNORE_INVALID_PROPERTY_STRING, propName, e);
+ logger.warn(IGNORE_INVALID_PROPERTY_STRING, IntegrityMonitorProperties.FAILED_COUNTER_THRESHOLD, e);
}
}
missedCycles += 1;
if (missedCycles >= failedCounterThreshold && !alarmExists) {
logger.debug("Forward progress not detected for resource {}. Setting state to disable failed.",
- resourceName);
+ resourceName);
if (!(stateManager.getOpState()).equals(StateManagement.DISABLED)) {
// Note: The refreshStateAudit will make redundant
// calls
return;
}
if (!stateManager.getStandbyStatus().equals(StateManagement.NULL_VALUE)
- && stateManager.getStandbyStatus() != null
- && !stateManager.getStandbyStatus().equals(StateManagement.PROVIDING_SERVICE)) {
+ && stateManager.getStandbyStatus() != null
+ && !stateManager.getStandbyStatus().equals(StateManagement.PROVIDING_SERVICE)) {
logger.debug("IntegrityMonitor.stateAudit(): NOT PROVIDING_SERVICE. returning");
return;
}
// Check if each forwardprogressentity entry is current
for (ForwardProgressEntity fpe : fpList) {
- // If the this is my ForwardProgressEntity, continue
+ // If this is my ForwardProgressEntity, continue
if (fpe.getResourceName().equals(IntegrityMonitor.resourceName)) {
continue;
}
long diffMs = date.getTime() - fpe.getLastUpdated().getTime();
if (logger.isDebugEnabled()) {
logger.debug("IntegrityMonitor.executeStateAudit(): resource = {}, diffMs = {}", fpe.getResourceName(),
- diffMs);
+ diffMs);
}
// Threshold for a stale entry
long staleMs = maxFpcUpdateIntervalMs;
if (logger.isDebugEnabled()) {
logger.debug("IntegrityMonitor.executeStateAudit(): resource = {}, staleMs = {}", fpe.getResourceName(),
- staleMs);
+ staleMs);
}
if (diffMs > staleMs) {
// ForwardProgress is stale. Disable it
// Start a transaction
logger.debug("IntegrityMonitor.executeStateAudit(): resource = {}, FPC is stale. Disabling it",
- fpe.getResourceName());
+ fpe.getResourceName());
EntityTransaction et = em.getTransaction();
et.begin();
StateManagementEntity sme = disableEntity(et, fpe);
* Disables the entity.
*
* @param entrans entity transaction
- * @param fpe entity of interest
+ * @param fpe entity of interest
* @return the corresponding state management entity
*/
private StateManagementEntity disableEntity(EntityTransaction entrans, ForwardProgressEntity fpe) {
try {
// query if StateManagement entry exists for fpe resource
TypedQuery<StateManagementEntity> query =
- em.createQuery("Select p from StateManagementEntity p where p.resourceName=:resource",
- StateManagementEntity.class);
+ em.createQuery("Select p from StateManagementEntity p where p.resourceName=:resource",
+ StateManagementEntity.class);
query.setParameter(LC_RESOURCE_STRING, fpe.getResourceName());
List<StateManagementEntity> smList =
- query.setLockMode(LockModeType.NONE).setFlushMode(FlushModeType.COMMIT).getResultList();
+ query.setLockMode(LockModeType.NONE).setFlushMode(FlushModeType.COMMIT).getResultList();
if (!smList.isEmpty()) {
// exists
sme = smList.get(0);
em.refresh(sme);
if (logger.isDebugEnabled()) {
logger.debug("IntegrityMonitor.executeStateAudit(): Found entry in StateManagementEntity table "
- + "for Resource={}", sme.getResourceName());
+ + "for Resource={}", sme.getResourceName());
}
} else {
String msg = "IntegrityMonitor.executeStateAudit(): " + fpe.getResourceName()
- + ": resource not found in state management entity database table";
+ + ": resource not found in state management entity database table";
logger.error("{}", msg);
}
synchronized (imFlushLock) {
} catch (Exception e) {
// log an error
logger.error("IntegrityMonitor.executeStateAudit(): {}: StateManagementEntity DB read failed with "
- + "exception: ", fpe.getResourceName(), e);
+ + "exception: ", fpe.getResourceName(), e);
synchronized (imFlushLock) {
if (entrans.isActive()) {
entrans.rollback();
private void disableEntity(String dep) {
try {
- // create instance of StateMangement class for dependent
+ // create instance of StateManagement class for dependent
var depStateManager = new StateManagement(emf, dep);
if (!depStateManager.getOpState().equals(StateManagement.DISABLED)) {
logger.debug("Forward progress not detected for dependent resource {}. Setting dependent's "
- + "state to disable failed.", dep);
+ + "state to disable failed.", dep);
depStateManager.disableFailed();
}
} catch (Exception e) {
private void disableFailed(StateManagementEntity sme) {
if (logger.isDebugEnabled()) {
logger.debug("IntegrityMonitor.executeStateAudit(): Changing OpStat = disabled for {}",
- sme.getResourceName());
+ sme.getResourceName());
}
try {
stateManager.disableFailed(sme.getResourceName());
long currTime = MonitorTime.getInstance().getMillis();
logger.debug("checkDependentHealth currTime - lastDependencyCheckTime = {}",
- currTime - lastDependencyCheckTime);
+ currTime - lastDependencyCheckTime);
if ((currTime - lastDependencyCheckTime) > checkDependencyIntervalMs) {
// execute dependency check and update this resource's state
/*
* This is a simple refresh audit which is periodically run to assure that the states and status attributes are
* aligned and notifications are sent to any listeners. It is possible for state/status to get out of sync and
- * notified systems to be out of synch due to database corruption (manual or otherwise) or because a node became
+ * notified systems to be out of sync due to database corruption (manual or otherwise) or because a node became
* isolated.
*
* When the operation (lock/unlock) is called, it will cause a re-evaluation of the state and send a notification to
* @param key the key
* @param asw <code>true</code> if all seems well for the key, <code>false</code> if all seems not well for the key
* @param msg message to add for the key
- * @throws AllSeemsWellException if an error occurs
*/
- public void allSeemsWell(String key, Boolean asw, String msg) throws AllSeemsWellException {
+ public void allSeemsWell(String key, Boolean asw, String msg) {
logger.debug("allSeemsWell entry: key = {}, asw = {}, msg = {}", key, asw, msg);
if (StringUtils.isEmpty(key)) {
}
/**
- * Converts the given value to milliseconds using the current {@link #propertyUnits}.
+ * Converts the given value to milliseconds using the current propertyUnits.
*
* @param value value to be converted, or -1
* @return the value, in milliseconds, or -1
// these methods may be overridden by junit tests
/**
- * Indicates that the {@link FpManager#run()} method has started. This method simply returns.
+ * Indicates that the {FpManager#run()} method has started. This method simply returns.
*
* @throws InterruptedException can be interrupted
*/
import jakarta.persistence.GeneratedValue;
import jakarta.persistence.GenerationType;
import jakarta.persistence.Id;
-import jakarta.persistence.NamedQueries;
import jakarta.persistence.NamedQuery;
import jakarta.persistence.PrePersist;
import jakarta.persistence.PreUpdate;
@Entity
@Table(name = "ImTestEntity")
-@NamedQueries({
- @NamedQuery(name = " ImTestEntity.findAll", query = "SELECT e FROM ImTestEntity e "),
- @NamedQuery(name = "ImTestEntity.deleteAll", query = "DELETE FROM ImTestEntity WHERE 1=1")
-})
-
+@NamedQuery(name = " ImTestEntity.findAll", query = "SELECT e FROM ImTestEntity e ")
+@NamedQuery(name = "ImTestEntity.deleteAll", query = "DELETE FROM ImTestEntity WHERE 1=1")
@Getter
@Setter
@NoArgsConstructor
<class>org.onap.policy.common.im.jpa.ResourceRegistrationEntity</class>
<shared-cache-mode>NONE</shared-cache-mode>
<properties>
- <property name="javax.persistence.schema-generation.database.action" value="drop-and-create" />
+ <property name="jakarta.persistence.schema-generation.database.action" value="drop-and-create" />
</properties>
</persistence-unit>
private static final String SITE_B_PAP2 = "siteB_pap2";
private static final String ENABLE_NO_DEPENDENCY_MSG =
- "\n\nsm.enableNoDependency()\nAdminState = {}\nOpState() = {}\n"
- + "AvailStatus = {}\nStandbyStatus = {}\n";
+ """
- private static Logger logger = LoggerFactory.getLogger(IntegrityMonitorTest.class);
+
+ sm.enableNoDependency()
+ AdminState = {}
+ OpState() = {}
+ AvailStatus = {}
+ StandbyStatus = {}
+ """;
+
+ private static final Logger logger = LoggerFactory.getLogger(IntegrityMonitorTest.class);
/**
* Number of monitor cycles it takes between dependency health checks.
im = makeMonitor(resourceName, myProp);
logger.debug(
- "\n\ntestSanityJmx state after creating new im\n"
- + "AdminState = {}\nOpState() = {}\nAvailStatus = {}\nStandbyStatus = {}\n",
+ """
+
+
+ testSanityJmx state after creating new im
+ AdminState = {}
+ OpState() = {}
+ AvailStatus = {}
+ StandbyStatus = {}
+ """,
im.getStateManager().getAdminState(), im.getStateManager().getOpState(),
im.getStateManager().getAvailStatus(), im.getStateManager().getStandbyStatus());
im = makeMonitor(resourceName, myProp);
sm = im.getStateManager();
logger.debug(
- "\n\ntestSanityJmx restart with AdminState=locked"
- + "\nAdminState = {}\nOpState() = {}\nAvailStatus = {}\nStandbyStatus = {}\n",
+ """
+
+
+ testSanityJmx restart with AdminState=locked
+ AdminState = {}
+ OpState() = {}
+ AvailStatus = {}
+ StandbyStatus = {}
+ """,
sm.getAdminState(), sm.getOpState(), sm.getAvailStatus(), sm.getStandbyStatus());
assertEquals(StateManagement.LOCKED, sm.getAdminState());
im = makeMonitor(resourceName, myProp);
sm = im.getStateManager();
logger.debug(
- "\n\ntestSanityJmx restart with AdminState=unlocked\n"
- + "AdminState = {}\nOpState() = {}\nAvailStatus = {}\nStandbyStatus = {}\n",
+ """
+
+
+ testSanityJmx restart with AdminState=unlocked
+ AdminState = {}
+ OpState() = {}
+ AvailStatus = {}
+ StandbyStatus = {}
+ """,
sm.getAdminState(), sm.getOpState(), sm.getAvailStatus(), sm.getStandbyStatus());
assertEquals(StateManagement.UNLOCKED, sm.getAdminState());
em.refresh(sme);
logger.debug(
- "??? -- Retrieve StateManagementEntity from database --\nsme.getResourceName() = {}\n"
- + "sme.getAdminState() = {}\nsme.getOpState() = {}\nsme.getAvailStatus() = {}\n"
- + "sme.getStandbyStatus() = {}",
+ """
+ ??? -- Retrieve StateManagementEntity from database --
+ sme.getResourceName() = {}
+ sme.getAdminState() = {}
+ sme.getOpState() = {}
+ sme.getAvailStatus() = {}
+ sme.getStandbyStatus() = {}""",
sme.getResourceName(), sme.getAdminState(), sme.getOpState(), sme.getAvailStatus(),
sme.getStandbyStatus());
sme1 = (StateManagementEntity) resourceList1.get(0);
em.refresh(sme1);
logger.debug(
- "??? -- Retrieve StateManagementEntity from database --\nsme1.getResourceName() = {}\n"
- + "sme1.getAdminState() = {}\nsme1.getOpState() = {}\nsme1.getAvailStatus() = {}\n"
- + "sme1.getStandbyStatus() = {}",
+ """
+ ??? -- Retrieve StateManagementEntity from database --
+ sme1.getResourceName() = {}
+ sme1.getAdminState() = {}
+ sme1.getOpState() = {}
+ sme1.getAvailStatus() = {}
+ sme1.getStandbyStatus() = {}""",
sme1.getResourceName(), sme1.getAdminState(), sme1.getOpState(), sme1.getAvailStatus(),
sme1.getStandbyStatus());
StateManagementEntity tmpSme = (StateManagementEntity) mySme;
em.refresh(tmpSme);
logger.debug(
- "\n ResourceName: {}\n AdminState: {}\n OpState: {}"
- + "\n AvailStatus: {}\n StandbyStatus: {}",
+ """
+
+ ResourceName: {}
+ AdminState: {}
+ OpState: {}
+ AvailStatus: {}
+ StandbyStatus: {}\
+ """,
tmpSme.getResourceName(), tmpSme.getAdminState(), tmpSme.getOpState(), tmpSme.getAvailStatus(),
tmpSme.getStandbyStatus());
}
StateManagementEntity tmpSme = (StateManagementEntity) mySme;
em.refresh(tmpSme);
logger.debug(
- "\n ResourceName: {}" + "\n AdminState: {}" + "\n OpState: {}"
- + "\n AvailStatus: {}" + "\n StandbyStatus: {}",
+ """
+
+ ResourceName: {}
+ AdminState: {}
+ OpState: {}
+ AvailStatus: {}
+ StandbyStatus: {}\
+ """,
tmpSme.getResourceName(), tmpSme.getAdminState(), tmpSme.getOpState(), tmpSme.getAvailStatus(),
tmpSme.getStandbyStatus());
}
em.refresh(tmpSme);
logger.debug(
- "\n (restart4) ResourceName: {}" + "\n AdminState: {}" + "\n OpState: {}"
- + "\n AvailStatus: {}" + "\n StandbyStatus: {}",
+ """
+
+ (restart4) ResourceName: {}
+ AdminState: {}
+ OpState: {}
+ AvailStatus: {}
+ StandbyStatus: {}\
+ """,
tmpSme.getResourceName(), tmpSme.getAdminState(), tmpSme.getOpState(), tmpSme.getAvailStatus(),
tmpSme.getStandbyStatus());
}
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
- <version>32.1.1-jre</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
* ONAP
* ================================================================================
* Copyright (C) 2017-2021 AT&T Intellectual Property. All rights reserved.
- * Copyright (C) 2022 Nordix Foundation.
+ * Modifications Copyright (C) 2022-2023 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Objects;
import java.util.Properties;
import lombok.Getter;
import org.onap.policy.common.capabilities.Startable;
* This implementation of the Topic Endpoint Manager, proxies operations to the appropriate
* implementation(s).
*/
+@Getter
class TopicEndpointProxy implements TopicEndpoint {
/**
* Logger.
/**
* Is this element locked boolean.
*/
- @Getter
private volatile boolean locked = false;
/**
* Is this element alive boolean.
*/
- @Getter
private volatile boolean alive = false;
@Override
@Override
public List<Topic> addTopics(TopicParameterGroup params) {
List<TopicParameters> sinks =
- (params.getTopicSinks() != null ? params.getTopicSinks() : Collections.emptyList());
+ (params.getTopicSinks() != null ? params.getTopicSinks() : Collections.emptyList());
List<TopicParameters> sources =
- (params.getTopicSources() != null ? params.getTopicSources() : Collections.emptyList());
+ (params.getTopicSources() != null ? params.getTopicSources() : Collections.emptyList());
List<Topic> topics = new ArrayList<>(sinks.size() + sources.size());
topics.addAll(addTopicSources(sources));
break;
default:
logger.debug("Unknown source type {} for topic: {}", param.getTopicCommInfrastructure(),
- param.getTopic());
+ param.getTopic());
break;
}
}
break;
default:
logger.debug("Unknown sink type {} for topic: {}", param.getTopicCommInfrastructure(),
- param.getTopic());
+ param.getTopic());
break;
}
}
}
final List<TopicSource> sources = new ArrayList<>();
- for (final String topic : topicNames) {
+
+ topicNames.forEach(topic -> {
try {
- final TopicSource uebSource = this.getUebTopicSource(topic);
- if (uebSource != null) {
- sources.add(uebSource);
- }
+ sources.add(Objects.requireNonNull(this.getUebTopicSource(topic)));
} catch (final Exception e) {
logger.debug("No UEB source for topic: {}", topic, e);
}
try {
- final TopicSource dmaapSource = this.getDmaapTopicSource(topic);
- if (dmaapSource != null) {
- sources.add(dmaapSource);
- }
+ sources.add(Objects.requireNonNull(this.getDmaapTopicSource(topic)));
} catch (final Exception e) {
logger.debug("No DMAAP source for topic: {}", topic, e);
}
try {
- final TopicSource kafkaSource = this.getKafkaTopicSource(topic);
- if (kafkaSource != null) {
- sources.add(kafkaSource);
- }
+ sources.add(Objects.requireNonNull(this.getKafkaTopicSource(topic)));
} catch (final Exception e) {
logger.debug("No KAFKA source for topic: {}", topic, e);
}
try {
- final TopicSource noopSource = this.getNoopTopicSource(topic);
- if (noopSource != null) {
- sources.add(noopSource);
- }
+ sources.add(Objects.requireNonNull(this.getNoopTopicSource(topic)));
} catch (final Exception e) {
logger.debug("No NOOP source for topic: {}", topic, e);
}
- }
+ });
+
return sources;
}
final List<TopicSink> sinks = new ArrayList<>();
for (final String topic : topicNames) {
try {
- final TopicSink uebSink = this.getUebTopicSink(topic);
- if (uebSink != null) {
- sinks.add(uebSink);
- }
+ sinks.add(Objects.requireNonNull(this.getUebTopicSink(topic)));
} catch (final Exception e) {
logger.debug("No UEB sink for topic: {}", topic, e);
}
try {
- final TopicSink dmaapSink = this.getDmaapTopicSink(topic);
- if (dmaapSink != null) {
- sinks.add(dmaapSink);
- }
+ sinks.add(Objects.requireNonNull(this.getDmaapTopicSink(topic)));
} catch (final Exception e) {
logger.debug("No DMAAP sink for topic: {}", topic, e);
}
try {
- final TopicSink kafkaSink = this.getKafkaTopicSink(topic);
- if (kafkaSink != null) {
- sinks.add(kafkaSink);
- }
+ sinks.add(Objects.requireNonNull(this.getKafkaTopicSink(topic)));
} catch (final Exception e) {
logger.debug("No KAFKA sink for topic: {}", topic, e);
}
try {
- final TopicSink noopSink = this.getNoopTopicSink(topic);
- if (noopSink != null) {
- sinks.add(noopSink);
- }
+ sinks.add(Objects.requireNonNull(this.getNoopTopicSink(topic)));
} catch (final Exception e) {
logger.debug("No NOOP sink for topic: {}", topic, e);
}
@Override
public List<TopicSink> getTopicSinks(String topicName) {
if (topicName == null) {
- throw parmException(null);
+ throw paramException(null);
}
final List<TopicSink> sinks = new ArrayList<>();
public TopicSource getTopicSource(Topic.CommInfrastructure commType, String topicName) {
if (commType == null) {
- throw parmException(topicName);
+ throw paramException(topicName);
}
if (topicName == null) {
- throw parmException(null);
+ throw paramException(null);
}
- switch (commType) {
- case UEB:
- return this.getUebTopicSource(topicName);
- case DMAAP:
- return this.getDmaapTopicSource(topicName);
- case KAFKA:
- return this.getKafkaTopicSource(topicName);
- case NOOP:
- return this.getNoopTopicSource(topicName);
- default:
- throw new UnsupportedOperationException("Unsupported " + commType.name());
- }
+ return switch (commType) {
+ case UEB -> this.getUebTopicSource(topicName);
+ case DMAAP -> this.getDmaapTopicSource(topicName);
+ case KAFKA -> this.getKafkaTopicSource(topicName);
+ case NOOP -> this.getNoopTopicSource(topicName);
+ default -> throw new UnsupportedOperationException("Unsupported " + commType.name());
+ };
}
@Override
public TopicSink getTopicSink(Topic.CommInfrastructure commType, String topicName) {
if (commType == null) {
- throw parmException(topicName);
+ throw paramException(topicName);
}
if (topicName == null) {
- throw parmException(null);
+ throw paramException(null);
}
- switch (commType) {
- case UEB:
- return this.getUebTopicSink(topicName);
- case DMAAP:
- return this.getDmaapTopicSink(topicName);
- case KAFKA:
- return this.getKafkaTopicSink(topicName);
- case NOOP:
- return this.getNoopTopicSink(topicName);
- default:
- throw new UnsupportedOperationException("Unsupported " + commType.name());
- }
+ return switch (commType) {
+ case UEB -> this.getUebTopicSink(topicName);
+ case DMAAP -> this.getDmaapTopicSink(topicName);
+ case KAFKA -> this.getKafkaTopicSink(topicName);
+ case NOOP -> this.getNoopTopicSink(topicName);
+ default -> throw new UnsupportedOperationException("Unsupported " + commType.name());
+ };
}
@Override
return NoopTopicFactories.getSinkFactory().get(topicName);
}
- private IllegalArgumentException parmException(String topicName) {
+ private IllegalArgumentException paramException(String topicName) {
return new IllegalArgumentException(
"Invalid parameter: a communication infrastructure required to fetch " + topicName);
}
* Copyright (C) 2017-2021 AT&T Intellectual Property. All rights reserved.
* Modifications Copyright (C) 2018 Samsung Electronics Co., Ltd.
* Modifications Copyright (C) 2020,2023 Bell Canada. All rights reserved.
- * Copyright (C) 2022 Nordix Foundation.
+ * Modifications Copyright (C) 2022-2023 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
import java.security.GeneralSecurityException;
import java.time.Duration;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
+import org.jetbrains.annotations.NotNull;
import org.onap.dmaap.mr.client.MRClientFactory;
import org.onap.dmaap.mr.client.impl.MRConsumerImpl;
import org.onap.dmaap.mr.client.impl.MRConsumerImpl.MRConsumerImplBuilder;
/**
* Consumer that handles fetch() failures by sleeping.
*/
- public abstract static class FetchingBusConsumer implements BusConsumer {
- private static Logger logger = LoggerFactory.getLogger(FetchingBusConsumer.class);
+ abstract class FetchingBusConsumer implements BusConsumer {
+ private static final Logger logger = LoggerFactory.getLogger(FetchingBusConsumer.class);
/**
* Fetch timeout.
/**
* Cambria Consumer Wrapper.
* BusTopicParam object contains the following parameters
- * servers messaging bus hosts.
- * topic topic
- * apiKey API Key
- * apiSecret API Secret
- * consumerGroup Consumer Group
- * consumerInstance Consumer Instance
- * fetchTimeout Fetch Timeout
- * fetchLimit Fetch Limit
+ * servers - messaging bus hosts.
+ * topic - topic for messages
+ * apiKey - API Key
+ * apiSecret - API Secret
+ * consumerGroup - Consumer Group
+ * consumerInstance - Consumer Instance
+ * fetchTimeout - Fetch Timeout
+ * fetchLimit - Fetch Limit
*
* @param busTopicParams - The parameters for the bus topic
- * @throws GeneralSecurityException - Security exception
- * @throws MalformedURLException - Malformed URL exception
*/
public CambriaConsumerWrapper(BusTopicParams busTopicParams) {
super(busTopicParams);
this.builder = new CambriaClientBuilders.ConsumerBuilder();
builder.knownAs(busTopicParams.getConsumerGroup(), busTopicParams.getConsumerInstance())
- .usingHosts(busTopicParams.getServers()).onTopic(busTopicParams.getTopic())
- .waitAtServer(fetchTimeout).receivingAtMost(busTopicParams.getFetchLimit());
+ .usingHosts(busTopicParams.getServers()).onTopic(busTopicParams.getTopic())
+ .waitAtServer(fetchTimeout).receivingAtMost(busTopicParams.getFetchLimit());
// Set read timeout to fetch timeout + 30 seconds (TBD: this should be configurable)
builder.withSocketTimeout(fetchTimeout + 30000);
/**
* Kafka based consumer.
*/
- public static class KafkaConsumerWrapper extends FetchingBusConsumer {
+ class KafkaConsumerWrapper extends FetchingBusConsumer {
/**
* logger.
*/
- private static Logger logger = LoggerFactory.getLogger(KafkaConsumerWrapper.class);
+ private static final Logger logger = LoggerFactory.getLogger(KafkaConsumerWrapper.class);
private static final String KEY_DESERIALIZER = "org.apache.kafka.common.serialization.StringDeserializer";
/**
* Kafka Consumer Wrapper.
- * BusTopicParam object contains the following parameters
- * servers messaging bus hosts.
- * topic topic
+ * BusTopicParam - object contains the following parameters
+ * servers - messaging bus hosts.
+ * topic - topic
*
* @param busTopicParams - The parameters for the bus topic
- * @throws GeneralSecurityException - Security exception
- * @throws MalformedURLException - Malformed URL exception
*/
- public KafkaConsumerWrapper(BusTopicParams busTopicParams) throws MalformedURLException {
+ public KafkaConsumerWrapper(BusTopicParams busTopicParams) {
super(busTopicParams);
if (busTopicParams.isTopicInvalid()) {
//Setup Properties for consumer
kafkaProps = new Properties();
kafkaProps.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
- busTopicParams.getServers().get(0));
+ busTopicParams.getServers().get(0));
if (busTopicParams.isAdditionalPropsValid()) {
- for (Map.Entry<String, String> entry : busTopicParams.getAdditionalProps().entrySet()) {
- kafkaProps.put(entry.getKey(), entry.getValue());
- }
+ kafkaProps.putAll(busTopicParams.getAdditionalProps());
}
if (kafkaProps.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG) == null) {
}
consumer = new KafkaConsumer<>(kafkaProps);
//Subscribe to the topic
- consumer.subscribe(Arrays.asList(busTopicParams.getTopic()));
+ consumer.subscribe(List.of(busTopicParams.getTopic()));
}
@Override
- public Iterable<String> fetch() throws IOException {
+ public Iterable<String> fetch() {
ConsumerRecords<String, String> records = this.consumer.poll(Duration.ofMillis(fetchTimeout));
if (records == null || records.count() <= 0) {
return Collections.emptyList();
consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
}
} catch (Exception e) {
- logger.error("{}: cannot fetch because of {}", this, e.getMessage());
+ logger.error("{}: cannot fetch, throwing exception after sleep...", this);
sleepAfterFetchFailure();
throw e;
}
/**
* logger.
*/
- private static Logger logger = LoggerFactory.getLogger(DmaapConsumerWrapper.class);
+ private static final Logger logger = LoggerFactory.getLogger(DmaapConsumerWrapper.class);
/**
* Name of the "protocol" property.
/**
* MR Consumer Wrapper.
*
- * <p>servers messaging bus hosts
- * topic topic
- * apiKey API Key
- * apiSecret API Secret
- * username AAF Login
- * password AAF Password
- * consumerGroup Consumer Group
- * consumerInstance Consumer Instance
- * fetchTimeout Fetch Timeout
- * fetchLimit Fetch Limit
+ * <p>servers - messaging bus hosts
+ * topic - topic
+ * apiKey - API Key
+ * apiSecret - API Secret
+ * username - AAF Login
+ * password - AAF Password
+ * consumerGroup - Consumer Group
+ * consumerInstance - Consumer Instance
+ * fetchTimeout - Fetch Timeout
+ * fetchLimit - Fetch Limit
*
* @param busTopicParams contains above listed attributes
* @throws MalformedURLException URL should be valid
}
this.consumer = new MRConsumerImplBuilder()
- .setHostPart(busTopicParams.getServers())
- .setTopic(busTopicParams.getTopic())
- .setConsumerGroup(busTopicParams.getConsumerGroup())
- .setConsumerId(busTopicParams.getConsumerInstance())
- .setTimeoutMs(busTopicParams.getFetchTimeout())
- .setLimit(busTopicParams.getFetchLimit())
- .setApiKey(busTopicParams.getApiKey())
- .setApiSecret(busTopicParams.getApiSecret())
- .createMRConsumerImpl();
+ .setHostPart(busTopicParams.getServers())
+ .setTopic(busTopicParams.getTopic())
+ .setConsumerGroup(busTopicParams.getConsumerGroup())
+ .setConsumerId(busTopicParams.getConsumerInstance())
+ .setTimeoutMs(busTopicParams.getFetchTimeout())
+ .setLimit(busTopicParams.getFetchLimit())
+ .setApiKey(busTopicParams.getApiKey())
+ .setApiSecret(busTopicParams.getApiSecret())
+ .createMRConsumerImpl();
this.consumer.setUsername(busTopicParams.getUserName());
this.consumer.setPassword(busTopicParams.getPassword());
}
@Override
- public Iterable<String> fetch() throws IOException {
+ public Iterable<String> fetch() {
final MRConsumerResponse response = this.consumer.fetchWithReturnConsumerResponse();
if (response == null) {
logger.warn("{}: DMaaP NULL response received", this);
return new ArrayList<>();
} else {
logger.debug("DMaaP consumer received {} : {}", response.getResponseCode(),
- response.getResponseMessage());
+ response.getResponseMessage());
if (!"200".equals(response.getResponseCode())) {
logger.error("DMaaP consumer received: {} : {}", response.getResponseCode(),
- response.getResponseMessage());
+ response.getResponseMessage());
sleepAfterFetchFailure();
@Override
public String toString() {
return "DmaapConsumerWrapper [" + "consumer.getAuthDate()=" + consumer.getAuthDate()
- + ", consumer.getAuthKey()=" + consumer.getAuthKey() + ", consumer.getHost()=" + consumer.getHost()
- + ", consumer.getProtocolFlag()=" + consumer.getProtocolFlag() + ", consumer.getUsername()="
- + consumer.getUsername() + "]";
+ + ", consumer.getAuthKey()=" + consumer.getAuthKey() + ", consumer.getHost()=" + consumer.getHost()
+ + ", consumer.getProtocolFlag()=" + consumer.getProtocolFlag() + ", consumer.getUsername()="
+ + consumer.getUsername() + "]";
}
}
/**
* MR based consumer.
*/
- public static class DmaapAafConsumerWrapper extends DmaapConsumerWrapper {
+ class DmaapAafConsumerWrapper extends DmaapConsumerWrapper {
- private static Logger logger = LoggerFactory.getLogger(DmaapAafConsumerWrapper.class);
-
- private final Properties props;
+ private static final Logger logger = LoggerFactory.getLogger(DmaapAafConsumerWrapper.class);
/**
* BusTopicParams contain the following parameters.
* MR Consumer Wrapper.
*
* <p>servers messaging bus hosts
- * topic topic
- * apiKey API Key
- * apiSecret API Secret
- * aafLogin AAF Login
- * aafPassword AAF Password
- * consumerGroup Consumer Group
- * consumerInstance Consumer Instance
- * fetchTimeout Fetch Timeout
- * fetchLimit Fetch Limit
+ * topic - topic
+ * apiKey - API Key
+ * apiSecret - API Secret
+ * aafLogin - AAF Login
+ * aafPassword - AAF Password
+ * consumerGroup - Consumer Group
+ * consumerInstance - Consumer Instance
+ * fetchTimeout - Fetch Timeout
+ * fetchLimit - Fetch Limit
*
* @param busTopicParams contains above listed params
* @throws MalformedURLException URL should be valid
this.consumer.setProtocolFlag(ProtocolTypeConstants.AAF_AUTH.getValue());
- props = new Properties();
+ Properties props = new Properties();
if (busTopicParams.isUseHttps()) {
props.setProperty(PROTOCOL_PROP, "https");
final MRConsumerImpl consumer = this.consumer;
return "DmaapConsumerWrapper [" + "consumer.getAuthDate()=" + consumer.getAuthDate()
- + ", consumer.getAuthKey()=" + consumer.getAuthKey() + ", consumer.getHost()=" + consumer.getHost()
- + ", consumer.getProtocolFlag()=" + consumer.getProtocolFlag() + ", consumer.getUsername()="
- + consumer.getUsername() + "]";
+ + ", consumer.getAuthKey()=" + consumer.getAuthKey() + ", consumer.getHost()=" + consumer.getHost()
+ + ", consumer.getProtocolFlag()=" + consumer.getProtocolFlag() + ", consumer.getUsername()="
+ + consumer.getUsername() + "]";
}
}
- public static class DmaapDmeConsumerWrapper extends DmaapConsumerWrapper {
-
- private static Logger logger = LoggerFactory.getLogger(DmaapDmeConsumerWrapper.class);
+ class DmaapDmeConsumerWrapper extends DmaapConsumerWrapper {
- private final Properties props;
+ private static final Logger logger = LoggerFactory.getLogger(DmaapDmeConsumerWrapper.class);
/**
* Constructor.
*
- * @param busTopicParams topic paramters
- *
+ * @param busTopicParams topic parameters
* @throws MalformedURLException must provide a valid URL
*/
public DmaapDmeConsumerWrapper(BusTopicParams busTopicParams) throws MalformedURLException {
final String dme2RouteOffer = (busTopicParams.isAdditionalPropsValid()
- ? busTopicParams.getAdditionalProps().get(
- PolicyEndPointProperties.DME2_ROUTE_OFFER_PROPERTY)
- : null);
+ ? busTopicParams.getAdditionalProps().get(
+ PolicyEndPointProperties.DME2_ROUTE_OFFER_PROPERTY)
+ : null);
- if (busTopicParams.isEnvironmentInvalid()) {
- throw parmException(busTopicParams.getTopic(),
- PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ENVIRONMENT_SUFFIX);
- }
- if (busTopicParams.isAftEnvironmentInvalid()) {
- throw parmException(busTopicParams.getTopic(),
- PolicyEndPointProperties.PROPERTY_DMAAP_DME2_AFT_ENVIRONMENT_SUFFIX);
- }
- if (busTopicParams.isLatitudeInvalid()) {
- throw parmException(busTopicParams.getTopic(),
- PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LATITUDE_SUFFIX);
- }
- if (busTopicParams.isLongitudeInvalid()) {
- throw parmException(busTopicParams.getTopic(),
- PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LONGITUDE_SUFFIX);
- }
+ BusHelper.validateBusTopicParams(busTopicParams, PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS);
if ((busTopicParams.isPartnerInvalid())
- && StringUtils.isBlank(dme2RouteOffer)) {
+ && StringUtils.isBlank(dme2RouteOffer)) {
throw new IllegalArgumentException(
- "Must provide at least " + PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
- + "." + busTopicParams.getTopic()
- + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_PARTNER_SUFFIX + " or "
- + PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
- + busTopicParams.getTopic()
- + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUTE_OFFER_SUFFIX + " for DME2");
+ "Must provide at least " + PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
+ + "." + busTopicParams.getTopic()
+ + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_PARTNER_SUFFIX + " or "
+ + PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
+ + busTopicParams.getTopic()
+ + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUTE_OFFER_SUFFIX + " for DME2");
}
final String serviceName = busTopicParams.getServers().get(0);
this.consumer.setUsername(busTopicParams.getUserName());
this.consumer.setPassword(busTopicParams.getPassword());
- props = new Properties();
+ Properties props = getProperties(busTopicParams, serviceName, dme2RouteOffer);
+
+ MRClientFactory.prop = props;
+ this.consumer.setProps(props);
+
+ logger.info("{}: CREATION", this);
+ }
+
+ @NotNull
+ private static Properties getProperties(BusTopicParams busTopicParams, String serviceName,
+ String dme2RouteOffer) {
+ Properties props = new Properties();
props.setProperty(PolicyEndPointProperties.DME2_SERVICE_NAME_PROPERTY, serviceName);
/* These are required, no defaults */
props.setProperty("topic", busTopicParams.getTopic());
- props.setProperty("Environment", busTopicParams.getEnvironment());
- props.setProperty("AFT_ENVIRONMENT", busTopicParams.getAftEnvironment());
+ BusHelper.setCommonProperties(busTopicParams, dme2RouteOffer, props);
- if (busTopicParams.getPartner() != null) {
- props.setProperty("Partner", busTopicParams.getPartner());
- }
- if (dme2RouteOffer != null) {
- props.setProperty(PolicyEndPointProperties.DME2_ROUTE_OFFER_PROPERTY, dme2RouteOffer);
- }
-
- props.setProperty("Latitude", busTopicParams.getLatitude());
- props.setProperty("Longitude", busTopicParams.getLongitude());
-
- /* These are optional, will default to these values if not set in additionalProps */
- props.setProperty("AFT_DME2_EP_READ_TIMEOUT_MS", "50000");
- props.setProperty("AFT_DME2_ROUNDTRIP_TIMEOUT_MS", "240000");
- props.setProperty("AFT_DME2_EP_CONN_TIMEOUT", "15000");
- props.setProperty("Version", "1.0");
- props.setProperty("SubContextPath", "/");
- props.setProperty("sessionstickinessrequired", "no");
-
- /* These should not change */
- props.setProperty("TransportType", "DME2");
props.setProperty("MethodType", "GET");
if (busTopicParams.isUseHttps()) {
props.setProperty("contenttype", "application/json");
if (busTopicParams.isAdditionalPropsValid()) {
- for (Map.Entry<String, String> entry : busTopicParams.getAdditionalProps().entrySet()) {
- props.put(entry.getKey(), entry.getValue());
- }
+ props.putAll(busTopicParams.getAdditionalProps());
}
-
- MRClientFactory.prop = props;
- this.consumer.setProps(props);
-
- logger.info("{}: CREATION", this);
- }
-
- private IllegalArgumentException parmException(String topic, String propnm) {
- return new IllegalArgumentException("Missing " + PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
- + topic + propnm + " property for DME2 in DMaaP");
-
+ return props;
}
}
}
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP POLICY
+ * ================================================================================
+ * Copyright (C) 2023 Nordix Foundation.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END============================================
+ * ===================================================================
+ *
+ */
+
+package org.onap.policy.common.endpoints.event.comm.bus.internal;
+
+import java.util.Properties;
+import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
+
+public class BusHelper {
+
+ private BusHelper() {
+ /* no constructor */
+ }
+
+ /**
+ * Complete the properties param with common fields for both BusConsumer and BusPublisher.
+ * @param busTopicParams topics
+ * @param dme2RouteOffer route
+ * @param props properties
+ */
+ public static void setCommonProperties(BusTopicParams busTopicParams, String dme2RouteOffer, Properties props) {
+ props.setProperty("Environment", busTopicParams.getEnvironment());
+ props.setProperty("AFT_ENVIRONMENT", busTopicParams.getAftEnvironment());
+
+ if (busTopicParams.getPartner() != null) {
+ props.setProperty("Partner", busTopicParams.getPartner());
+ }
+ if (dme2RouteOffer != null) {
+ props.setProperty(PolicyEndPointProperties.DME2_ROUTE_OFFER_PROPERTY, dme2RouteOffer);
+ }
+
+ props.setProperty("Latitude", busTopicParams.getLatitude());
+ props.setProperty("Longitude", busTopicParams.getLongitude());
+
+ /* These are optional, will default to these values if not set in additionalProps */
+ props.setProperty("AFT_DME2_EP_READ_TIMEOUT_MS", "50000");
+ props.setProperty("AFT_DME2_ROUNDTRIP_TIMEOUT_MS", "240000");
+ props.setProperty("AFT_DME2_EP_CONN_TIMEOUT", "15000");
+ props.setProperty("Version", "1.0");
+ props.setProperty("SubContextPath", "/");
+ props.setProperty("sessionstickinessrequired", "no");
+
+ /* These should not change */
+ props.setProperty("TransportType", "DME2");
+ }
+
+ /**
+ * Throws exception when any of the checks are invalid.
+ * @param busTopicParams topics
+ * @param topicType topic type (sink or source)
+ */
+ public static void validateBusTopicParams(BusTopicParams busTopicParams, String topicType) {
+ if (busTopicParams.isEnvironmentInvalid()) {
+ throw paramException(busTopicParams.getTopic(), topicType,
+ PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ENVIRONMENT_SUFFIX);
+ }
+ if (busTopicParams.isAftEnvironmentInvalid()) {
+ throw paramException(busTopicParams.getTopic(), topicType,
+ PolicyEndPointProperties.PROPERTY_DMAAP_DME2_AFT_ENVIRONMENT_SUFFIX);
+ }
+ if (busTopicParams.isLatitudeInvalid()) {
+ throw paramException(busTopicParams.getTopic(), topicType,
+ PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LATITUDE_SUFFIX);
+ }
+ if (busTopicParams.isLongitudeInvalid()) {
+ throw paramException(busTopicParams.getTopic(), topicType,
+ PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LONGITUDE_SUFFIX);
+ }
+ }
+
+ private static IllegalArgumentException paramException(String topic, String topicType, String propertyName) {
+ return new IllegalArgumentException("Missing " + topicType + "."
+ + topic + propertyName + " property for DME2 in DMaaP");
+
+ }
+}
public interface BusPublisher {
- public static final String NO_MESSAGE_PROVIDED = "No message provided";
+ String NO_MESSAGE_PROVIDED = "No message provided";
+ String LOG_CLOSE = "{}: CLOSE";
+ String LOG_CLOSE_FAILED = "{}: CLOSE FAILED";
/**
* sends a message.
* @return true if success, false otherwise
* @throws IllegalArgumentException if no message provided
*/
- public boolean send(String partitionId, String message);
+ boolean send(String partitionId, String message);
/**
* closes the publisher.
*/
- public void close();
+ void close();
/**
* Cambria based library publisher.
*/
- public static class CambriaPublisherWrapper implements BusPublisher {
+ class CambriaPublisherWrapper implements BusPublisher {
- private static Logger logger = LoggerFactory.getLogger(CambriaPublisherWrapper.class);
+ private static final Logger logger = LoggerFactory.getLogger(CambriaPublisherWrapper.class);
/**
* The actual Cambria publisher.
@Override
public void close() {
- logger.info("{}: CLOSE", this);
+ logger.info(LOG_CLOSE, this);
try {
this.publisher.close();
/**
* Kafka based library publisher.
*/
- public static class KafkaPublisherWrapper implements BusPublisher {
+ class KafkaPublisherWrapper implements BusPublisher {
- private static Logger logger = LoggerFactory.getLogger(KafkaPublisherWrapper.class);
+ private static final Logger logger = LoggerFactory.getLogger(KafkaPublisherWrapper.class);
private static final String KEY_SERIALIZER = "org.apache.kafka.common.serialization.StringSerializer";
- private String topic;
+ private final String topic;
/**
* Kafka publisher.
*/
- private Producer<String, String> producer;
+ private final Producer<String, String> producer;
protected Properties kafkaProps;
/**
kafkaProps = new Properties();
kafkaProps.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, busTopicParams.getServers().get(0));
if (busTopicParams.isAdditionalPropsValid()) {
- for (Map.Entry<String, String> entry : busTopicParams.getAdditionalProps().entrySet()) {
- kafkaProps.put(entry.getKey(), entry.getValue());
- }
+ kafkaProps.putAll(busTopicParams.getAdditionalProps());
}
if (kafkaProps.get(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG) == null) {
kafkaProps.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, KEY_SERIALIZER);
@Override
public void close() {
- logger.info("{}: CLOSE", this);
+ logger.info(LOG_CLOSE, this);
try {
this.producer.close();
/**
* DmaapClient library wrapper.
*/
- public abstract class DmaapPublisherWrapper implements BusPublisher {
+ abstract class DmaapPublisherWrapper implements BusPublisher {
- private static Logger logger = LoggerFactory.getLogger(DmaapPublisherWrapper.class);
+ private static final Logger logger = LoggerFactory.getLogger(DmaapPublisherWrapper.class);
/**
* MR based Publisher.
@Override
public void close() {
- logger.info("{}: CLOSE", this);
+ logger.info(LOG_CLOSE, this);
try {
this.publisher.close(1, TimeUnit.SECONDS);
} catch (InterruptedException e) {
- logger.warn("{}: CLOSE FAILED", this, e);
+ logger.warn(LOG_CLOSE_FAILED, this, e);
Thread.currentThread().interrupt();
} catch (Exception e) {
- logger.warn("{}: CLOSE FAILED", this, e);
+ logger.warn(LOG_CLOSE_FAILED, this, e);
}
}
/**
* DmaapClient library wrapper.
*/
- public static class DmaapAafPublisherWrapper extends DmaapPublisherWrapper {
+ class DmaapAafPublisherWrapper extends DmaapPublisherWrapper {
/**
* MR based Publisher.
*/
}
}
- public static class DmaapDmePublisherWrapper extends DmaapPublisherWrapper {
+ class DmaapDmePublisherWrapper extends DmaapPublisherWrapper {
/**
* Constructor.
String serviceName = busTopicParams.getServers().get(0);
/* These are required, no defaults */
- props.setProperty("Environment", busTopicParams.getEnvironment());
- props.setProperty("AFT_ENVIRONMENT", busTopicParams.getAftEnvironment());
-
props.setProperty(PolicyEndPointProperties.DME2_SERVICE_NAME_PROPERTY, serviceName);
- if (busTopicParams.getPartner() != null) {
- props.setProperty("Partner", busTopicParams.getPartner());
- }
- if (dme2RouteOffer != null) {
- props.setProperty(PolicyEndPointProperties.DME2_ROUTE_OFFER_PROPERTY, dme2RouteOffer);
- }
-
- props.setProperty("Latitude", busTopicParams.getLatitude());
- props.setProperty("Longitude", busTopicParams.getLongitude());
-
- // ServiceName also a default, found in additionalProps
-
- /* These are optional, will default to these values if not set in optionalProps */
- props.setProperty("AFT_DME2_EP_READ_TIMEOUT_MS", "50000");
- props.setProperty("AFT_DME2_ROUNDTRIP_TIMEOUT_MS", "240000");
- props.setProperty("AFT_DME2_EP_CONN_TIMEOUT", "15000");
- props.setProperty("Version", "1.0");
- props.setProperty("SubContextPath", "/");
- props.setProperty("sessionstickinessrequired", "no");
+ BusHelper.setCommonProperties(busTopicParams, dme2RouteOffer, props);
- /* These should not change */
- props.setProperty("TransportType", "DME2");
props.setProperty("MethodType", "POST");
if (busTopicParams.isAdditionalPropsValid()) {
}
private void validateParams(BusTopicParams busTopicParams, String dme2RouteOffer) {
- if (busTopicParams.isEnvironmentInvalid()) {
- throw parmException(busTopicParams.getTopic(),
- PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ENVIRONMENT_SUFFIX);
- }
- if (busTopicParams.isAftEnvironmentInvalid()) {
- throw parmException(busTopicParams.getTopic(),
- PolicyEndPointProperties.PROPERTY_DMAAP_DME2_AFT_ENVIRONMENT_SUFFIX);
- }
- if (busTopicParams.isLatitudeInvalid()) {
- throw parmException(busTopicParams.getTopic(),
- PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LATITUDE_SUFFIX);
- }
- if (busTopicParams.isLongitudeInvalid()) {
- throw parmException(busTopicParams.getTopic(),
- PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LONGITUDE_SUFFIX);
- }
+ BusHelper.validateBusTopicParams(busTopicParams, PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS);
if ((busTopicParams.isPartnerInvalid()) && StringUtils.isBlank(dme2RouteOffer)) {
throw new IllegalArgumentException("Must provide at least "
}
}
}
-
- private IllegalArgumentException parmException(String topic, String propnm) {
- return new IllegalArgumentException("Missing " + PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS + "."
- + topic + propnm + " property for DME2 in DMaaP");
-
- }
}
}
* ================================================================================
* Copyright (C) 2017-2021 AT&T Intellectual Property. All rights reserved.
* Modifications Copyright (C) 2020 Bell Canada. All rights reserved.
+ * Modifications Copyright (C) 2023 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
/**
* Logger.
*/
- private static Logger logger = LoggerFactory.getLogger(TopicBase.class);
+ private static final Logger logger = LoggerFactory.getLogger(TopicBase.class);
/**
* List of servers.
public interface HttpServletServerFactory {
/**
- * Builds an http or https rest server with support for servlets.
+ * Builds a http or https rest server with support for servlets.
*
* @param name name
* @param https use secured http over tls connection
boolean swagger, boolean managed);
/**
- * Builds an http rest server with support for servlets.
+ * Builds a http rest server with support for servlets.
*
* @param name name
* @param host binding host
List<HttpServletServer> build(Properties properties);
/**
- * Builds an http or https server to manage static resources.
+ * Builds a http or https server to manage static resources.
*
* @param name name
* @param https use secured http over tls connection
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThrows;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
}
@Test
- public void testFetchingBusConsumer() throws InterruptedException {
+ public void testFetchingBusConsumer() {
// should not be negative
var cons = new FetchingBusConsumerImpl(makeBuilder().fetchTimeout(-1).build());
assertThat(cons.getSleepTime()).isEqualTo(PolicyEndPointProperties.DEFAULT_TIMEOUT_MS_FETCH);
@Test
public void testCambriaConsumerWrapperClose() {
CambriaConsumerWrapper cons = new CambriaConsumerWrapper(builder.build());
- assertThatCode(() -> cons.close()).doesNotThrowAnyException();
+ assertThatCode(cons::close).doesNotThrowAnyException();
}
@Test
}
@Test
- public void testDmaapConsumerWrapper() throws Exception {
+ public void testDmaapConsumerWrapper() {
// verify that different wrappers can be built
assertThatCode(() -> new DmaapAafConsumerWrapper(makeBuilder().build())).doesNotThrowAnyException();
}
}
@Test
- public void testDmaapConsumerWrapperClose() throws Exception {
+ public void testDmaapConsumerWrapperClose() {
assertThatCode(() -> new DmaapAafConsumerWrapper(makeBuilder().build()).close()).doesNotThrowAnyException();
}
}
@Test
- public void testKafkaConsumerWrapper() throws Exception {
+ public void testKafkaConsumerWrapper() {
// verify that different wrappers can be built
assertThatCode(() -> new KafkaConsumerWrapper(makeKafkaBuilder().build())).doesNotThrowAnyException();
}
@Test(expected = IllegalArgumentException.class)
- public void testKafkaConsumerWrapper_InvalidTopic() throws Exception {
+ public void testKafkaConsumerWrapper_InvalidTopic() {
new KafkaConsumerWrapper(makeBuilder().topic(null).build());
}
- @Test(expected = java.lang.IllegalStateException.class)
- public void testKafkaConsumerWrapperFetch() throws Exception {
+ @Test
+ public void testKafkaConsumerWrapperFetch() {
//Setup Properties for consumer
Properties kafkaProps = new Properties();
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(kafkaProps);
kafka.consumer = consumer;
- assertFalse(kafka.fetch().iterator().hasNext());
+ assertThrows(java.lang.IllegalStateException.class, () -> kafka.fetch().iterator().hasNext());
consumer.close();
}
@Test
- public void testKafkaConsumerWrapperClose() throws Exception {
+ public void testKafkaConsumerWrapperClose() {
assertThatCode(() -> new KafkaConsumerWrapper(makeKafkaBuilder().build()).close()).doesNotThrowAnyException();
}
@Test
- public void testKafkaConsumerWrapperToString() throws Exception {
+ public void testKafkaConsumerWrapperToString() {
assertNotNull(new KafkaConsumerWrapper(makeKafkaBuilder().build()) {}.toString());
}
}
@Override
- public Iterable<String> fetch() throws IOException {
+ public Iterable<String> fetch() {
return null;
}
}
* ONAP POLICY
* ================================================================================
* Copyright (C) 2021 AT&T Intellectual Property. All right reserved.
+ * Modifications Copyright (C) 2023 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
return FilterReply.DENY;
}
- if (event == null || event.getMarker() == null) {
+ if (event == null || event.getMarkerList() == null) {
return FilterReply.DENY;
}
- if (event.getMarker().equals(marker)) {
+ if (event.getMarkerList().stream().anyMatch(mk -> mk.equals(marker))) {
return FilterReply.ACCEPT;
} else {
return FilterReply.DENY;
/*-
* ============LICENSE_START=======================================================
* Copyright (C) 2018 Ericsson. All rights reserved.
- * Modifications Copyright (C) 2020 Nordix Foundation.
+ * Modifications Copyright (C) 2020, 2023 Nordix Foundation.
* Modifications Copyright (C) 2020-2021 AT&T Intellectual Property. All rights reserved.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.Enumeration;
+import java.util.Objects;
import java.util.Set;
import java.util.TreeSet;
import java.util.jar.JarEntry;
*/
public static String getResourceAsString(final String resourceName) {
// Get the resource as a stream, we'll convert it to a string then
- final InputStream resourceStream = getResourceAsStream(resourceName);
- if (resourceStream == null) {
- return null;
- }
-
// Read the stream contents, closing when done
- try (var streamCloser = resourceStream) {
+ try (var resourceStream = getResourceAsStream(resourceName)) {
+ if (resourceStream == null) {
+ return null;
+ }
return IOUtils.toString(resourceStream, StandardCharsets.UTF_8);
} catch (final IOException e) {
LOGGER.debug("error reading resource stream {}", resourceName, e);
// Check if the resource exists
if (urlToResource == null) {
// No resource found
- LOGGER.debug("cound not find resource \"{}\" : ", resourceName);
+ LOGGER.debug("could not find resource \"{}\" : ", resourceName);
return null;
}
* Read the list of entries in a resource directory.
*
* @param resourceDirectoryName the name of the resource directory
- * @return the list of entries
+ * @return a set of entries
*/
public static Set<String> getDirectoryContents(final String resourceDirectoryName) {
// Find the location of the resource, is it in a Jar or on the local file system?
*
* @param localResourceDirectoryUrl the local resource file URL
* @param resourceDirectoryName the name of the resource directory
- * @return a list of the directory contents
+ * @return a set of the directory contents
*/
public static Set<String> getDirectoryContentsLocal(final URL localResourceDirectoryUrl,
final String resourceDirectoryName) {
}
Set<String> localDirectorySet = new TreeSet<>();
- for (File localDirectoryEntry : localDirectory.listFiles()) {
+ for (File localDirectoryEntry : Objects.requireNonNull(localDirectory.listFiles())) {
if (localDirectoryEntry.isDirectory()) {
localDirectorySet
.add(resourceDirectoryName + File.separator + localDirectoryEntry.getName() + File.separator);
*
* @param jarResourceDirectoryUrl the name of the resource directory in the jar
* @param resourceDirectoryName the name of the resource directory
- * @return a list of the directory contents
+ * @return a set of the directory contents
*/
public static Set<String> getDirectoryContentsJar(final URL jarResourceDirectoryUrl,
final String resourceDirectoryName) {
Set<String> localDirectorySet = new TreeSet<>();
try (var jarFile = new JarFile(jarFileName)) {
- Enumeration<JarEntry> entries = jarFile.entries();
+ Enumeration<JarEntry> entries = jarFile.entries(); // NOSONAR
while (entries.hasMoreElements()) {
/*
Code Review / policy / common.git / commitdiff