diff options
Diffstat (limited to 'aai-core')
13 files changed, 560 insertions, 420 deletions
diff --git a/aai-core/pom.xml b/aai-core/pom.xml index bd759f1f..c091546b 100644 --- a/aai-core/pom.xml +++ b/aai-core/pom.xml @@ -388,6 +388,11 @@ limitations under the License. <artifactId>spring-jms</artifactId> </dependency> <dependency> + <groupId>javax.jms</groupId> + <artifactId>javax.jms-api</artifactId> + <version>2.0.1</version> + </dependency> + <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-core</artifactId> </dependency> diff --git a/aai-core/src/main/java/org/onap/aai/kafka/AAIKafkaEventJMSConsumer.java b/aai-core/src/main/java/org/onap/aai/kafka/AAIKafkaEventJMSConsumer.java index 731f3dfc..67f6842e 100644 --- a/aai-core/src/main/java/org/onap/aai/kafka/AAIKafkaEventJMSConsumer.java +++ b/aai-core/src/main/java/org/onap/aai/kafka/AAIKafkaEventJMSConsumer.java @@ -20,116 +20,114 @@ * ============LICENSE_END========================================================= */ - package org.onap.aai.kafka; - - import java.util.Map; - import java.util.Objects; - - import javax.jms.JMSException; - import javax.jms.Message; - import javax.jms.MessageListener; - import javax.jms.TextMessage; - - import org.json.JSONException; - import org.json.JSONObject; - import org.onap.aai.aailog.logs.AaiDmaapMetricLog; - import org.onap.aai.exceptions.AAIException; - import org.onap.aai.logging.AaiElsErrorCode; - import org.onap.aai.logging.ErrorLogHelper; - import org.slf4j.Logger; - import org.slf4j.LoggerFactory; - import org.slf4j.MDC; - import org.springframework.core.env.Environment; - import org.springframework.kafka.core.KafkaTemplate; - - public class AAIKafkaEventJMSConsumer implements MessageListener { - - private static final String EVENT_TOPIC = "event-topic"; - - private static final Logger LOGGER = LoggerFactory.getLogger(AAIKafkaEventJMSConsumer.class); - - private Environment environment; - private Map<String, String> mdcCopy; - private KafkaTemplate<String,String> kafkaTemplate; - - public AAIKafkaEventJMSConsumer(Environment environment,KafkaTemplate<String,String> kafkaTemplate) { - super(); - mdcCopy = MDC.getCopyOfContextMap(); - Objects.nonNull(environment); - this.environment = environment; - this.kafkaTemplate=kafkaTemplate; - } - - @Override - public void onMessage(Message message) { - - if (kafkaTemplate == null) { - return; - } - - String jsmMessageTxt = ""; - String aaiEvent = ""; - JSONObject aaiEventHeader; - JSONObject joPayload; - String transactionId = ""; - String serviceName = ""; - String eventName = ""; - String aaiElsErrorCode = AaiElsErrorCode.SUCCESS; - String errorDescription = ""; - - if (mdcCopy != null) { - MDC.setContextMap(mdcCopy); - } - - if (message instanceof TextMessage) { - AaiDmaapMetricLog metricLog = new AaiDmaapMetricLog(); - try { - jsmMessageTxt = ((TextMessage) message).getText(); - JSONObject jo = new JSONObject(jsmMessageTxt); - if (jo.has("aaiEventPayload")) { - joPayload = jo.getJSONObject("aaiEventPayload"); - aaiEvent = joPayload.toString(); - } else { - return; - } - if (jo.getString(EVENT_TOPIC) != null) { - eventName = jo.getString(EVENT_TOPIC); - } - if (joPayload.has("event-header")) { - try { - aaiEventHeader = joPayload.getJSONObject("event-header"); - if (aaiEventHeader.has("id")) { - transactionId = aaiEventHeader.get("id").toString(); - } - if (aaiEventHeader.has("entity-link")) { - serviceName = aaiEventHeader.get("entity-link").toString(); - } - } catch (JSONException jexc) { - // ignore, this is just used for logging - } - } - metricLog.pre(eventName, aaiEvent, transactionId, serviceName); - - - if ("AAI-EVENT".equals(eventName)) { - // restTemplate.exchange(baseUrl + endpoint, HttpMethod.POST, httpEntity, String.class); - kafkaTemplate.send(eventName,aaiEvent); - - } else { - LOGGER.error(String.format("%s|Event Topic invalid.", eventName)); - } - } catch (JMSException | JSONException e) { - aaiElsErrorCode = AaiElsErrorCode.DATA_ERROR; - errorDescription = e.getMessage(); - ErrorLogHelper.logException(new AAIException("AAI_7350")); - } catch (Exception e) { - aaiElsErrorCode = AaiElsErrorCode.AVAILABILITY_TIMEOUT_ERROR; - errorDescription = e.getMessage(); - ErrorLogHelper.logException(new AAIException("AAI_7304", jsmMessageTxt)); - } finally { - metricLog.post(aaiElsErrorCode, errorDescription); - } - } - } - } -
\ No newline at end of file +package org.onap.aai.kafka; + +import java.util.Map; + +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.MessageListener; +import javax.jms.TextMessage; + +import org.json.JSONException; +import org.json.JSONObject; +import org.onap.aai.aailog.logs.AaiDmaapMetricLog; +import org.onap.aai.exceptions.AAIException; +import org.onap.aai.logging.AaiElsErrorCode; +import org.onap.aai.logging.ErrorLogHelper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.slf4j.MDC; +import org.springframework.kafka.core.KafkaTemplate; + + +public class AAIKafkaEventJMSConsumer implements MessageListener { + + private static final String EVENT_TOPIC = "event-topic"; + + private static final Logger LOGGER = LoggerFactory.getLogger(AAIKafkaEventJMSConsumer.class); + + private Map<String, String> mdcCopy; + private final KafkaTemplate<String, String> kafkaTemplate; + + public AAIKafkaEventJMSConsumer(KafkaTemplate<String, String> kafkaTemplate) { + super(); + mdcCopy = MDC.getCopyOfContextMap(); + this.kafkaTemplate = kafkaTemplate; + } + + @Override + public void onMessage(Message message) { + + if (kafkaTemplate == null) { + return; + } + + String jmsMessageText = ""; + String aaiEvent = ""; + JSONObject aaiEventHeader; + JSONObject aaiEventPayload; + String transactionId = ""; + String serviceName = ""; + String topicName = ""; + String aaiElsErrorCode = AaiElsErrorCode.SUCCESS; + String errorDescription = ""; + + if (mdcCopy != null) { + MDC.setContextMap(mdcCopy); + } + + if (message instanceof TextMessage) { + AaiDmaapMetricLog metricLog = new AaiDmaapMetricLog(); + try { + jmsMessageText = ((TextMessage) message).getText(); + JSONObject jsonObject = new JSONObject(jmsMessageText); + if (jsonObject.has("aaiEventPayload")) { + aaiEventPayload = jsonObject.getJSONObject("aaiEventPayload"); + aaiEvent = aaiEventPayload.toString(); + } else { + return; + } + if (jsonObject.getString(EVENT_TOPIC) != null) { + topicName = jsonObject.getString(EVENT_TOPIC); + } + if (aaiEventPayload.has("event-header")) { + try { + aaiEventHeader = aaiEventPayload.getJSONObject("event-header"); + if (aaiEventHeader.has("id")) { + transactionId = aaiEventHeader.get("id").toString(); + } + if (aaiEventHeader.has("entity-link")) { + serviceName = aaiEventHeader.get("entity-link").toString(); + } + } catch (JSONException jexc) { + // ignore, this is just used for logging + } + } + metricLog.pre(topicName, aaiEvent, transactionId, serviceName); + + if ("AAI-EVENT".equals(topicName)) { + + kafkaTemplate.send(topicName, aaiEvent); + + } else { + LOGGER.error(String.format("%s|Event Topic invalid.", topicName)); + } + } catch (JMSException | JSONException e) { + aaiElsErrorCode = AaiElsErrorCode.DATA_ERROR; + errorDescription = e.getMessage(); + ErrorLogHelper.logException(new AAIException("AAI_7350")); + } catch (Exception e) { + e.printStackTrace(); + // LOGGER.error(); + LOGGER.error(e.getMessage()); + aaiElsErrorCode = AaiElsErrorCode.AVAILABILITY_TIMEOUT_ERROR; + errorDescription = e.getMessage(); + String errorMessage = String.format("Error processing message: %s, message payload: %s", e.getMessage(), jmsMessageText); + ErrorLogHelper.logException(new AAIException("AAI_7304", errorMessage)); + } finally { + metricLog.post(aaiElsErrorCode, errorDescription); + } + } + } +} diff --git a/aai-core/src/main/java/org/onap/aai/kafka/AAIKafkaEventJMSProducer.java b/aai-core/src/main/java/org/onap/aai/kafka/AAIKafkaEventJMSProducer.java index 00cf677f..a46f2e99 100644 --- a/aai-core/src/main/java/org/onap/aai/kafka/AAIKafkaEventJMSProducer.java +++ b/aai-core/src/main/java/org/onap/aai/kafka/AAIKafkaEventJMSProducer.java @@ -22,44 +22,28 @@ package org.onap.aai.kafka; -import org.apache.activemq.ActiveMQConnectionFactory; -import org.apache.activemq.command.ActiveMQQueue; import org.json.JSONObject; import org.onap.aai.util.AAIConfig; -import org.springframework.jms.connection.CachingConnectionFactory; +import org.springframework.beans.factory.annotation.Value; import org.springframework.jms.core.JmsTemplate; +import org.springframework.stereotype.Service; +import lombok.RequiredArgsConstructor; + +@Service +@RequiredArgsConstructor public class AAIKafkaEventJMSProducer implements MessageProducer { - private JmsTemplate jmsTemplate; + @Value("${aai.events.enabled:true}") private boolean eventsEnabled; + private final JmsTemplate jmsTemplate; - public AAIKafkaEventJMSProducer() { - if ("true".equals(AAIConfig.get("aai.jms.enable", "true"))) { - this.jmsTemplate = new JmsTemplate(); - String activeMqTcpUrl = System.getProperty("activemq.tcp.url", "tcp://localhost:61547"); - this.jmsTemplate - .setConnectionFactory(new CachingConnectionFactory(new ActiveMQConnectionFactory(activeMqTcpUrl))); - this.jmsTemplate.setDefaultDestination(new ActiveMQQueue("IN_QUEUE")); + public void sendMessageToDefaultDestination(String msg) { + if (eventsEnabled) { + jmsTemplate.convertAndSend(msg); } } public void sendMessageToDefaultDestination(JSONObject finalJson) { - if (jmsTemplate != null) { - jmsTemplate.convertAndSend(finalJson.toString()); - CachingConnectionFactory ccf = (CachingConnectionFactory) this.jmsTemplate.getConnectionFactory(); - if (ccf != null) { - ccf.destroy(); - } - } - } - - public void sendMessageToDefaultDestination(String msg) { - if (jmsTemplate != null) { - jmsTemplate.convertAndSend(msg); - CachingConnectionFactory ccf = (CachingConnectionFactory) this.jmsTemplate.getConnectionFactory(); - if (ccf != null) { - ccf.destroy(); - } - } + sendMessageToDefaultDestination(finalJson.toString()); } } diff --git a/aai-core/src/main/java/org/onap/aai/util/StoreNotificationEvent.java b/aai-core/src/main/java/org/onap/aai/util/StoreNotificationEvent.java index 6f3e8883..127cf538 100644 --- a/aai-core/src/main/java/org/onap/aai/util/StoreNotificationEvent.java +++ b/aai-core/src/main/java/org/onap/aai/util/StoreNotificationEvent.java @@ -39,14 +39,18 @@ import org.onap.aai.kafka.AAIKafkaEventJMSProducer; import org.onap.aai.kafka.MessageProducer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.core.env.Environment; +import org.springframework.jms.core.JmsTemplate; public class StoreNotificationEvent { private static final Logger LOGGER = LoggerFactory.getLogger(StoreNotificationEvent.class); - private MessageProducer messageProducer; + @Autowired JmsTemplate jmsTemplate; + + private final MessageProducer messageProducer; private String fromAppId = ""; private String transId = ""; private final String transactionId; @@ -59,12 +63,12 @@ public class StoreNotificationEvent { * Instantiates a new store notification event. */ public StoreNotificationEvent(String transactionId, String sourceOfTruth) { - this.messageProducer = new AAIKafkaEventJMSProducer(); + this.messageProducer = new AAIKafkaEventJMSProducer(jmsTemplate); this.transactionId = transactionId; this.sourceOfTruth = sourceOfTruth; } - public StoreNotificationEvent(AAIKafkaEventJMSProducer producer, String transactionId, String sourceOfTruth) { + public StoreNotificationEvent(MessageProducer producer, String transactionId, String sourceOfTruth) { this.messageProducer = producer; this.transactionId = transactionId; this.sourceOfTruth = sourceOfTruth; diff --git a/aai-core/src/main/java/org/onap/aai/util/delta/DeltaEvents.java b/aai-core/src/main/java/org/onap/aai/util/delta/DeltaEvents.java index 6fbc297b..b2821b04 100644 --- a/aai-core/src/main/java/org/onap/aai/util/delta/DeltaEvents.java +++ b/aai-core/src/main/java/org/onap/aai/util/delta/DeltaEvents.java @@ -20,7 +20,6 @@ package org.onap.aai.util.delta; -import com.google.gson.*; import java.text.DateFormat; import java.text.SimpleDateFormat; @@ -28,38 +27,35 @@ import java.util.Date; import java.util.Map; import org.onap.aai.db.props.AAIProperties; -import org.onap.aai.kafka.AAIKafkaEventJMSProducer; import org.onap.aai.kafka.MessageProducer; import org.onap.aai.util.AAIConfig; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; -public class DeltaEvents { +import com.google.gson.FieldNamingPolicy; +import com.google.gson.Gson; +import com.google.gson.GsonBuilder; +import com.google.gson.JsonObject; - private static final Logger LOGGER = LoggerFactory.getLogger(DeltaEvents.class); +public class DeltaEvents { private static final Gson gson = new GsonBuilder().setFieldNamingPolicy(FieldNamingPolicy.LOWER_CASE_WITH_DASHES).create(); + private static final String eventVersion = "v1"; - private String transId; - private String sourceName; - private String eventVersion = "v1"; - private String schemaVersion; - private Map<String, ObjectDelta> objectDeltas; + private final String transId; + private final String sourceName; + private final String schemaVersion; + private final Map<String, ObjectDelta> objectDeltas; - private MessageProducer messageProducer; + @Autowired private MessageProducer messageProducer; public DeltaEvents(String transId, String sourceName, String schemaVersion, Map<String, ObjectDelta> objectDeltas) { - this(transId, sourceName, schemaVersion, objectDeltas, new AAIKafkaEventJMSProducer()); - } - - public DeltaEvents(String transId, String sourceName, String schemaVersion, Map<String, ObjectDelta> objectDeltas, - MessageProducer messageProducer) { - this.transId = transId; - this.sourceName = sourceName; - this.schemaVersion = schemaVersion; - this.objectDeltas = objectDeltas; - this.messageProducer = messageProducer; + this.transId = transId; + this.sourceName = sourceName; + this.schemaVersion = schemaVersion; + this.objectDeltas = objectDeltas; } public boolean triggerEvents() { @@ -98,7 +94,7 @@ public class DeltaEvents { header.addProperty("source-name", this.sourceName); header.addProperty("domain", this.getDomain()); header.addProperty("event-type", this.getEventType()); - header.addProperty("event-version", this.eventVersion); + header.addProperty("event-version", eventVersion); header.addProperty("schema-version", this.schemaVersion); header.addProperty("action", first.getAction().toString()); header.addProperty("entity-type", this.getEntityType(first)); @@ -126,7 +122,7 @@ public class DeltaEvents { /** * Given Long timestamp convert to format YYYYMMdd-HH:mm:ss:SSS - * + * * @param timestamp milliseconds since epoc * @return long timestamp in format YYYYMMdd-HH:mm:ss:SSS */ diff --git a/aai-core/src/main/java/org/onap/aai/web/KafkaConfig.java b/aai-core/src/main/java/org/onap/aai/web/KafkaConfig.java index 71ae5b6b..05e4768d 100644 --- a/aai-core/src/main/java/org/onap/aai/web/KafkaConfig.java +++ b/aai-core/src/main/java/org/onap/aai/web/KafkaConfig.java @@ -18,158 +18,167 @@ * ============LICENSE_END========================================================= */ - package org.onap.aai.web; - - import java.util.HashMap; - import java.util.Map; - - import javax.annotation.PostConstruct; - - import org.apache.activemq.ActiveMQConnectionFactory; - import org.apache.activemq.broker.BrokerService; - import org.apache.activemq.command.ActiveMQQueue; - import org.apache.kafka.clients.producer.ProducerConfig; +package org.onap.aai.web; + +import java.util.HashMap; +import java.util.Map; + +import javax.annotation.PostConstruct; +import javax.jms.ConnectionFactory; +import javax.jms.MessageListener; +import javax.jms.Queue; + +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.command.ActiveMQQueue; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.onap.aai.introspection.LoaderFactory; import org.onap.aai.kafka.AAIKafkaEventJMSConsumer; import org.onap.aai.kafka.AAIKafkaEventJMSProducer; +import org.onap.aai.kafka.MessageProducer; +import org.onap.aai.rest.notification.NotificationService; import org.slf4j.Logger; - import org.slf4j.LoggerFactory; - import org.springframework.beans.factory.annotation.Autowired; - import org.springframework.beans.factory.annotation.Value; - import org.springframework.context.ApplicationContext; - import org.springframework.context.annotation.Bean; - import org.springframework.context.annotation.Configuration; - import org.springframework.context.annotation.Profile; - import org.springframework.jms.connection.CachingConnectionFactory; - import org.springframework.jms.core.JmsTemplate; - import org.springframework.jms.listener.DefaultMessageListenerContainer; - import org.springframework.kafka.core.DefaultKafkaProducerFactory; - import org.springframework.kafka.core.KafkaTemplate; - import org.springframework.kafka.core.ProducerFactory; - - @Profile("kafka") - @Configuration - public class KafkaConfig { - - @Autowired - private ApplicationContext ctx; - - - @Value("${jms.bind.address}") - private String bindAddress; - - @Value("${spring.kafka.producer.bootstrap-servers}") - private String bootstrapServers; - - @Value("${spring.kafka.producer.properties.security.protocol}") - private String securityProtocol; - - @Value("${spring.kafka.producer.properties.sasl.mechanism}") - private String saslMechanism; - - @Value("${spring.kafka.producer.properties.sasl.jaas.config}") - private String saslJaasConfig; - - @Value("${spring.kafka.producer.retries}") - private Integer retries; - - private static final Logger logger = LoggerFactory.getLogger(KafkaConfig.class); - - @PostConstruct - public void init() { - System.setProperty("activemq.tcp.url", bindAddress); - } - - @Bean(destroyMethod = "stop") - public BrokerService brokerService() throws Exception { - - BrokerService broker = new BrokerService(); - broker.addConnector(bindAddress); - broker.setPersistent(false); - broker.setUseJmx(false); - broker.setSchedulerSupport(false); - broker.start(); - - return broker; - } - - @Bean(name = "connectionFactory") - public ActiveMQConnectionFactory activeMQConnectionFactory() { - return new ActiveMQConnectionFactory(bindAddress); - } - - @Bean - public CachingConnectionFactory cachingConnectionFactory() { - return new CachingConnectionFactory(activeMQConnectionFactory()); - } - - @Bean(name = "destinationQueue") - public ActiveMQQueue activeMQQueue() { - return new ActiveMQQueue("IN_QUEUE"); - } - - @Bean - public JmsTemplate jmsTemplate() { - JmsTemplate jmsTemplate = new JmsTemplate(); - - jmsTemplate.setConnectionFactory(activeMQConnectionFactory()); - jmsTemplate.setDefaultDestination(activeMQQueue()); - - return jmsTemplate; - } - - @Bean - public AAIKafkaEventJMSProducer jmsProducer() { - return new AAIKafkaEventJMSProducer(); - } - - @Bean(name = "jmsConsumer") - public AAIKafkaEventJMSConsumer jmsConsumer() throws Exception { - return new AAIKafkaEventJMSConsumer(ctx.getEnvironment(),kafkaTemplate()); - } - - @Bean - public DefaultMessageListenerContainer defaultMessageListenerContainer() throws Exception { - - DefaultMessageListenerContainer messageListenerContainer = new DefaultMessageListenerContainer(); - - messageListenerContainer.setConnectionFactory(cachingConnectionFactory()); - messageListenerContainer.setDestinationName("IN_QUEUE"); - messageListenerContainer.setMessageListener(jmsConsumer()); - - return messageListenerContainer; - } - - @Bean - public ProducerFactory<String, String> producerFactory() throws Exception { - Map<String, Object> props = new HashMap<>(); - if(bootstrapServers == null){ - logger.error("Environment Variable " + bootstrapServers + " is missing"); - throw new Exception("Environment Variable " + bootstrapServers + " is missing"); - } - else{ - props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); - } - if(saslJaasConfig == null){ - logger.info("Not using any authentication for kafka interaction"); - } - else{ - logger.info("Using authentication provided by kafka interaction"); - // Strimzi Kafka security properties - props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); - props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); - props.put("security.protocol", securityProtocol); - props.put("sasl.mechanism", saslMechanism); - props.put("sasl.jaas.config", saslJaasConfig); - props.put(ProducerConfig.RETRIES_CONFIG, Integer.toString(retries)); - props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION,"5"); - } - - return new DefaultKafkaProducerFactory<>(props); - } - - @Bean - public KafkaTemplate<String, String> kafkaTemplate() throws Exception { - return new KafkaTemplate<>(producerFactory()); - } - } -
\ No newline at end of file +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Profile; +import org.springframework.jms.connection.CachingConnectionFactory; +import org.springframework.jms.core.JmsTemplate; +import org.springframework.jms.listener.DefaultMessageListenerContainer; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; + +@Profile("kafka") +@Configuration +public class KafkaConfig { + + @Value("${jms.bind.address}") + private String bindAddress; + + @Value("${spring.kafka.producer.bootstrap-servers}") + private String bootstrapServers; + + @Value("${spring.kafka.producer.properties.security.protocol}") + private String securityProtocol; + + @Value("${spring.kafka.producer.properties.sasl.mechanism}") + private String saslMechanism; + + @Value("${spring.kafka.producer.properties.sasl.jaas.config}") + private String saslJaasConfig; + + @Value("${spring.kafka.producer.retries}") + private String retries; + + private static final Logger logger = LoggerFactory.getLogger(KafkaConfig.class); + + @PostConstruct + public void init() { + System.setProperty("activemq.tcp.url", bindAddress); + } + + @Bean(destroyMethod = "stop") + public BrokerService brokerService() throws Exception { + + BrokerService broker = new BrokerService(); + broker.addConnector(bindAddress); + broker.setPersistent(false); + broker.setUseJmx(false); + broker.setSchedulerSupport(false); + broker.start(); + + return broker; + } + + @ConditionalOnMissingBean + @Bean(name = "connectionFactory") + public ConnectionFactory activeMQConnectionFactory() { + return new ActiveMQConnectionFactory(bindAddress); + } + + @Bean + @ConditionalOnMissingBean + public CachingConnectionFactory cachingConnectionFactory(ConnectionFactory targetConnectionFactory) { + return new CachingConnectionFactory(targetConnectionFactory); + } + + @Bean(name = "destinationQueue") + public Queue activeMQQueue() { + return new ActiveMQQueue("IN_QUEUE"); + } + + @Bean + public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory, Queue queue) { + JmsTemplate jmsTemplate = new JmsTemplate(); + + jmsTemplate.setConnectionFactory(connectionFactory); + jmsTemplate.setDefaultDestination(queue); + + return jmsTemplate; + } + + @Bean(name = "jmsConsumer") + public MessageListener jmsConsumer(KafkaTemplate<String, String> kafkaTemplate) throws Exception { + return new AAIKafkaEventJMSConsumer(kafkaTemplate); + } + + @Bean + public DefaultMessageListenerContainer defaultMessageListenerContainer(ConnectionFactory connectionFactory, MessageListener messageListener) + throws Exception { + + DefaultMessageListenerContainer messageListenerContainer = new DefaultMessageListenerContainer(); + + messageListenerContainer.setConnectionFactory(connectionFactory); + messageListenerContainer.setDestinationName("IN_QUEUE"); + messageListenerContainer.setMessageListener(messageListener); + + return messageListenerContainer; + } + + @Bean + public ProducerFactory<String, String> producerFactory() throws Exception { + Map<String, Object> props = new HashMap<>(); + if (bootstrapServers == null) { + logger.error("Environment Variable " + bootstrapServers + " is missing"); + throw new Exception("Environment Variable " + bootstrapServers + " is missing"); + } else { + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + } + props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + props.put(ProducerConfig.RETRIES_CONFIG, retries); + props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5"); + + if (saslJaasConfig == null) { + logger.info("Not using any authentication for kafka interaction"); + } else { + logger.info("Using authentication provided by kafka interaction"); + // Strimzi Kafka security properties + props.put("security.protocol", securityProtocol); + props.put("sasl.mechanism", saslMechanism); + props.put("sasl.jaas.config", saslJaasConfig); + } + + return new DefaultKafkaProducerFactory<>(props); + } + + @Bean + public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> producerFactory) throws Exception { + return new KafkaTemplate<>(producerFactory); + } + + @Bean + public MessageProducer messageProducer(JmsTemplate jmsTemplate) { + return new AAIKafkaEventJMSProducer(jmsTemplate); + } + + @Bean + public NotificationService notificationService(LoaderFactory loaderFactory, + @Value("${schema.uri.base.path}") String basePath, + @Value("${delta.events.enabled:false}") boolean isDeltaEventsEnabled) { + return new NotificationService(loaderFactory, basePath, isDeltaEventsEnabled); + } +} diff --git a/aai-core/src/main/resources/logback.xml b/aai-core/src/main/resources/logback.xml index fc66b32e..ba5b3de8 100644 --- a/aai-core/src/main/resources/logback.xml +++ b/aai-core/src/main/resources/logback.xml @@ -166,7 +166,7 @@ <pattern>${eelfTransLogPattern}</pattern> </encoder> </appender> - + <appender name="asynctranslog" class="ch.qos.logback.classic.AsyncAppender"> <queueSize>1000</queueSize> <includeCallerData>true</includeCallerData> @@ -258,7 +258,7 @@ <logger name="ajsc.UserDefinedBeansDefService" level="WARN" /> <logger name="ajsc.LoggingConfigurationService" level="WARN" /> - <!-- AJSC related loggers (DME2 Registration, csi logging, restlet, servlet + <!-- AJSC related loggers (DME2 Registration, csi logging, restlet, servlet logging) --> <logger name="org.codehaus.groovy" level="WARN" /> <logger name="com.att.scamper" level="WARN" /> @@ -282,7 +282,7 @@ <logger name="org.apache.coyote" level="WARN" /> <logger name="org.apache.jasper" level="WARN" /> - <!-- Camel Related Loggers (including restlet/servlet/jaxrs/cxf logging. + <!-- Camel Related Loggers (including restlet/servlet/jaxrs/cxf logging. May aid in troubleshooting) --> <logger name="org.apache.camel" level="WARN" /> <logger name="org.apache.cxf" level="WARN" /> @@ -366,9 +366,7 @@ </logger> <logger name="org.onap.aai.kafka" level="DEBUG" additivity="false"> - <appender-ref ref="kafkaAAIEventConsumer" /> - <appender-ref ref="kafkaAAIEventConsumerDebug" /> - <appender-ref ref="kafkaAAIEventConsumerMetric" /> + <appender-ref ref="STDOUT" /> </logger> <logger name="org.apache" level="OFF" /> diff --git a/aai-core/src/test/java/org/onap/aai/AAISetup.java b/aai-core/src/test/java/org/onap/aai/AAISetup.java index 16f21ff4..08a0e91b 100644 --- a/aai-core/src/test/java/org/onap/aai/AAISetup.java +++ b/aai-core/src/test/java/org/onap/aai/AAISetup.java @@ -40,6 +40,7 @@ import org.onap.aai.setup.AAIConfigTranslator; import org.onap.aai.setup.SchemaVersion; import org.onap.aai.setup.SchemaVersions; import org.onap.aai.util.AAIConstants; +import org.onap.aai.web.KafkaConfig; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.test.context.ContextConfiguration; @@ -52,7 +53,8 @@ import org.springframework.test.context.web.WebAppConfiguration; @ContextConfiguration( classes = {ConfigConfiguration.class, AAIConfigTranslator.class, EdgeIngestor.class, EdgeSerializer.class, NodeIngestor.class, SpringContextAware.class, IntrospectionConfig.class, RestBeanConfig.class, - XmlFormatTransformerConfiguration.class, ValidationService.class, ValidationConfiguration.class, LoaderFactory.class, NotificationService.class}) + XmlFormatTransformerConfiguration.class, ValidationService.class, ValidationConfiguration.class, + KafkaConfig.class, LoaderFactory.class, NotificationService.class}) @TestPropertySource( properties = {"schema.uri.base.path = /aai", "schema.xsd.maxoccurs = 5000", "schema.translator.list=config", "schema.nodes.location=src/test/resources/onap/oxm", diff --git a/aai-core/src/test/java/org/onap/aai/kafka/AAIKafkaEventIntegrationTest.java b/aai-core/src/test/java/org/onap/aai/kafka/AAIKafkaEventIntegrationTest.java new file mode 100644 index 00000000..c10260da --- /dev/null +++ b/aai-core/src/test/java/org/onap/aai/kafka/AAIKafkaEventIntegrationTest.java @@ -0,0 +1,97 @@ +/** + * ============LICENSE_START======================================================= + * org.onap.aai + * ================================================================================ + * Copyright © 2024 Deutsche Telekom. All rights reserved. + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END========================================================= + */ +package org.onap.aai.kafka; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.junit.jupiter.api.Assertions.assertFalse; + +import java.time.Duration; +import java.util.Collections; +import java.util.List; + +import javax.ws.rs.core.Response; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.TopicPartition; +import org.json.JSONObject; +import org.junit.Test; +import org.mockito.Mock; +import org.onap.aai.AAISetup; +import org.onap.aai.PayloadUtil; +import org.onap.aai.restcore.HttpMethod; +import org.skyscreamer.jsonassert.JSONAssert; +import org.skyscreamer.jsonassert.JSONCompareMode; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Import; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.test.context.ActiveProfiles; +import org.springframework.test.context.TestPropertySource; + +import lombok.extern.slf4j.Slf4j; + +@Slf4j +@ActiveProfiles("kafka") +@Import(KafkaTestConfiguration.class) +@EmbeddedKafka(partitions = 1, topics = { "AAI-EVENT" }) +@TestPropertySource( + properties = { + "jms.bind.address=tcp://localhost:61647", + "aai.events.enabled=true", + "spring.kafka.producer.retries=0", + "spring.kafka.producer.properties.sasl.jaas.config=#{null}", + "spring.kafka.producer.bootstrap-servers=${spring.embedded.kafka.brokers}" + }) +public class AAIKafkaEventIntegrationTest extends AAISetup { + + @Mock + private KafkaTemplate<String, String> kafkaTemplate; + + @Autowired + MessageProducer messageProducer; + + @Autowired + private ConsumerFactory<String, String> consumerFactory; + + @Test + public void onMessage_shouldSendMessageToKafkaTopic_whenAAIEventReceived() + throws Exception { + Consumer<String, String> consumer = consumerFactory.createConsumer(); + + consumer.subscribe(Collections.singletonList("AAI-EVENT")); + + String payload = PayloadUtil.getResourcePayload("aai-event.json"); + String expectedResponse = PayloadUtil.getExpectedPayload("aai-event.json"); + messageProducer.sendMessageToDefaultDestination(payload); + + ConsumerRecords<String, String> consumerRecords = KafkaTestUtils.getRecords(consumer, 10000); + assertFalse(consumerRecords.isEmpty()); + consumerRecords.forEach(consumerRecord -> { + JSONAssert.assertEquals(expectedResponse, consumerRecord.value(), JSONCompareMode.NON_EXTENSIBLE); + }); + } + +} diff --git a/aai-core/src/test/java/org/onap/aai/kafka/AAIKafkaEventJMSConsumerTest.java b/aai-core/src/test/java/org/onap/aai/kafka/AAIKafkaEventJMSConsumerTest.java deleted file mode 100644 index c72499c4..00000000 --- a/aai-core/src/test/java/org/onap/aai/kafka/AAIKafkaEventJMSConsumerTest.java +++ /dev/null @@ -1,89 +0,0 @@ -package org.onap.aai.kafka; - -import static org.mockito.ArgumentMatchers.anyString; -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; - -import javax.jms.TextMessage; - -import org.junit.Before; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.mockito.Mock; -import org.mockito.junit.MockitoJUnitRunner; -import org.onap.aai.PayloadUtil; -import org.springframework.core.env.Environment; -import org.springframework.kafka.core.KafkaTemplate; -import org.springframework.kafka.test.context.EmbeddedKafka; -import org.springframework.test.util.ReflectionTestUtils; - -@RunWith(MockitoJUnitRunner.class) -@EmbeddedKafka(partitions = 1, brokerProperties = { "listeners=PLAINTEXT://localhost:9092", "port=9092" }) -public class AAIKafkaEventJMSConsumerTest { - - @Mock - private Environment environment; - - @Mock - private KafkaTemplate<String,String> kafkaTemplate; - - private AAIKafkaEventJMSConsumer aaiKafkaEventJMSConsumer; - - @Before - public void setUp(){ - aaiKafkaEventJMSConsumer = new AAIKafkaEventJMSConsumer(environment,kafkaTemplate); - } - - @Test - public void onMessage_shouldSendMessageToKafkaTopic_whenAAIEventReceived() - throws Exception - { - TextMessage mockTextMessage = mock(TextMessage.class); - String payload = PayloadUtil.getResourcePayload("aai-event.json"); - - when(mockTextMessage.getText()).thenReturn(payload); - aaiKafkaEventJMSConsumer.onMessage(mockTextMessage); - verify(kafkaTemplate, times(1)).send(eq("AAI-EVENT"), anyString()); - } - - @Test - public void onMessage_shouldNotSendMessageToKafkaTopic_whenInvalidEventReceived() throws Exception{ - TextMessage mockTextMessage = mock(TextMessage.class); - String payload = PayloadUtil.getResourcePayload("aai-invalid-event.json"); - when(mockTextMessage.getText()).thenReturn(payload); - aaiKafkaEventJMSConsumer.onMessage(mockTextMessage); - } - - - @Test - public void onMessage_shouldHandleJSONException() throws Exception { - // Arrange - AAIKafkaEventJMSConsumer consumer = new AAIKafkaEventJMSConsumer(null, kafkaTemplate); - TextMessage mockTextMessage = mock(TextMessage.class); - ReflectionTestUtils.setField(consumer, "kafkaTemplate", null); // Simulate null kafkaTemplate - - // Act - consumer.onMessage(mockTextMessage); - - // Assert - // Verify that exception is logged - } - - @Test - public void onMessage_shouldHandleGenericException() throws Exception { - // Arrange - AAIKafkaEventJMSConsumer consumer = new AAIKafkaEventJMSConsumer(null, kafkaTemplate); - TextMessage mockTextMessage = mock(TextMessage.class); - when(mockTextMessage.getText()).thenReturn("{\"event-topic\":\"AAI-EVENT\",\"aaiEventPayload\":{}}"); // Valid JSON but missing required fields - - // Act - consumer.onMessage(mockTextMessage); - - // Assert - // Verify that exception is logged - } - -} diff --git a/aai-core/src/test/java/org/onap/aai/kafka/KafkaTestConfiguration.java b/aai-core/src/test/java/org/onap/aai/kafka/KafkaTestConfiguration.java new file mode 100644 index 00000000..730699e6 --- /dev/null +++ b/aai-core/src/test/java/org/onap/aai/kafka/KafkaTestConfiguration.java @@ -0,0 +1,77 @@ +/** + * ============LICENSE_START======================================================= + * org.onap.aai + * ================================================================================ + * Copyright © 2024 Deutsche Telekom. All rights reserved. + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END========================================================= + */ +package org.onap.aai.kafka; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.kafka.clients.admin.AdminClientConfig; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.KafkaAdmin; +import org.springframework.test.context.TestPropertySource; + +@TestConfiguration +public class KafkaTestConfiguration { + + @Value("${spring.embedded.kafka.brokers}") private String bootstrapAddress; + + private String groupId = "test-consumer"; + + @Bean + public KafkaAdmin kafkaAdmin() { + Map<String, Object> configs = new HashMap<>(); + configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress); + return new KafkaAdmin(configs); + } + + @Bean + public ConsumerFactory<String, String> consumerFactory() { + Map<String, Object> props = new HashMap<>(); + props.put( + ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, + bootstrapAddress); + props.put( + ConsumerConfig.GROUP_ID_CONFIG, + groupId); + props.put( + ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, + StringDeserializer.class); + props.put( + ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, + StringDeserializer.class); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + return new DefaultKafkaConsumerFactory<>(props); + } + + @Bean + public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(ConsumerFactory<String, String> consumerFactory) { + + ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>(); + factory.setConsumerFactory(consumerFactory); + return factory; + } +} diff --git a/aai-core/src/test/resources/logback.xml b/aai-core/src/test/resources/logback.xml index 4c82c0bf..6acc77f0 100644 --- a/aai-core/src/test/resources/logback.xml +++ b/aai-core/src/test/resources/logback.xml @@ -163,7 +163,7 @@ <pattern>${eelfTransLogPattern}</pattern> </encoder> </appender> - + <appender name="asynctranslog" class="ch.qos.logback.classic.AsyncAppender"> <queueSize>1000</queueSize> <includeCallerData>true</includeCallerData> @@ -255,7 +255,7 @@ <logger name="ajsc.UserDefinedBeansDefService" level="WARN" /> <logger name="ajsc.LoggingConfigurationService" level="WARN" /> - <!-- AJSC related loggers (DME2 Registration, csi logging, restlet, servlet + <!-- AJSC related loggers (DME2 Registration, csi logging, restlet, servlet logging) --> <logger name="org.codehaus.groovy" level="WARN" /> <logger name="com.att.scamper" level="WARN" /> @@ -279,7 +279,7 @@ <logger name="org.apache.coyote" level="WARN" /> <logger name="org.apache.jasper" level="WARN" /> - <!-- Camel Related Loggers (including restlet/servlet/jaxrs/cxf logging. + <!-- Camel Related Loggers (including restlet/servlet/jaxrs/cxf logging. May aid in troubleshooting) --> <logger name="org.apache.camel" level="WARN" /> <logger name="org.apache.cxf" level="WARN" /> @@ -363,9 +363,7 @@ </logger> <logger name="org.onap.aai.kafka" level="DEBUG" additivity="false"> - <appender-ref ref="kafkaAAIEventConsumer" /> - <appender-ref ref="kafkaAAIEventConsumerDebug" /> - <appender-ref ref="kafkaAAIEventConsumerMetric" /> + <appender-ref ref="STDOUT" /> </logger> <logger name="org.apache" level="WARN" /> diff --git a/aai-core/src/test/resources/payloads/expected/aai-event.json b/aai-core/src/test/resources/payloads/expected/aai-event.json new file mode 100644 index 00000000..86c67992 --- /dev/null +++ b/aai-core/src/test/resources/payloads/expected/aai-event.json @@ -0,0 +1,61 @@ +{ + "cambria.partition": "AAI", + "event-header": { + "severity": "NORMAL", + "entity-type": "object-group", + "top-entity-type": "object-group", + "entity-link": "/aai/v28/common/object-groups/object-group/ric_cluster", + "event-type": "AAI-EVENT", + "domain": "dev", + "action": "UPDATE", + "sequence-number": "0", + "id": "23f12123-c326-48a7-b57e-e48746c295ea", + "source-name": "postman-api", + "version": "v28", + "timestamp": "20231207-12:14:44:757" + }, + "entity": { + "relationship-list": { + "relationship": [ + { + "related-to": "cell", + "relationship-data": [ + { + "relationship-value": "445611193273958916", + "relationship-key": "cell.cell-id" + } + ], + "related-link": "/aai/v28/network/cells/cell/445611193273958916", + "relationship-label": "org.onap.relationships.inventory.MemberOf", + "related-to-property": [ + { + "property-key": "cell.cell-name", + "property-value": "MY6885_M-Schwere-Reiter-Str-440460_GU2_84079913" + } + ] + }, + { + "related-to": "cell", + "relationship-data": [ + { + "relationship-value": "445611193272330241", + "relationship-key": "cell.cell-id" + } + ], + "related-link": "/aai/v28/network/cells/cell/445611193272330241", + "relationship-label": "org.onap.relationships.inventory.MemberOf", + "related-to-property": [ + { + "property-key": "cell.cell-name", + "property-value": "MY6885_M-Schwere-Reiter-Str-440460_GTC2_84003803" + } + ] + } + ] + }, + "group-name": "Urban", + "resource-version": "1701951284582", + "group-type": "cell", + "object-group-id": "ric_cluster" + } +} |