diff options
Diffstat (limited to 'catalog-be/src/main/java/org/openecomp/sdc/be/components/kafka/KafkaHandler.java')
-rw-r--r-- | catalog-be/src/main/java/org/openecomp/sdc/be/components/kafka/KafkaHandler.java | 138 |
1 files changed, 138 insertions, 0 deletions
diff --git a/catalog-be/src/main/java/org/openecomp/sdc/be/components/kafka/KafkaHandler.java b/catalog-be/src/main/java/org/openecomp/sdc/be/components/kafka/KafkaHandler.java new file mode 100644 index 0000000000..2a5590e72d --- /dev/null +++ b/catalog-be/src/main/java/org/openecomp/sdc/be/components/kafka/KafkaHandler.java @@ -0,0 +1,138 @@ +/*- + * ============LICENSE_START======================================================= + * SDC + * ================================================================================ + * Copyright (C) 2022 Nordix Foundation. 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.openecomp.sdc.be.components.kafka; + +import com.google.gson.Gson; +import com.google.gson.JsonSyntaxException; +import fj.data.Either; +import lombok.Getter; +import lombok.Setter; +import org.apache.http.HttpStatus; +import org.apache.kafka.common.KafkaException; +import org.openecomp.sdc.be.components.distribution.engine.CambriaErrorResponse; +import org.openecomp.sdc.be.components.distribution.engine.INotificationData; +import org.openecomp.sdc.be.config.BeEcompErrorManager; +import org.openecomp.sdc.be.config.ConfigurationManager; +import org.openecomp.sdc.be.config.DistributionEngineConfiguration; +import org.openecomp.sdc.be.distribution.api.client.CambriaOperationStatus; +import org.openecomp.sdc.common.log.wrappers.Logger; +import org.springframework.stereotype.Component; + +/** + * Utility class that provides a handler for Kafka interactions + */ +@Component +public class KafkaHandler { + + private static final Logger log = Logger.getLogger(KafkaHandler.class.getName()); + private final Gson gson = new Gson(); + + private SdcKafkaConsumer sdcKafkaConsumer; + + private SdcKafkaProducer sdcKafkaProducer; + + @Setter + private boolean isKafkaActive; + + private DistributionEngineConfiguration deConfiguration; + + public KafkaHandler(SdcKafkaConsumer sdcKafkaConsumer, SdcKafkaProducer sdcKafkaProducer, boolean isKafkaActive) { + this.sdcKafkaConsumer = sdcKafkaConsumer; + this.sdcKafkaProducer = sdcKafkaProducer; + this.isKafkaActive = isKafkaActive; + } + + public KafkaHandler() { + isKafkaActive = Boolean.parseBoolean(System.getenv().getOrDefault("USE_KAFKA", "false")); + deConfiguration = ConfigurationManager.getConfigurationManager().getDistributionEngineConfiguration(); + } + + /** + * @return a user configuration whether Kafka is active for this client + */ + public Boolean isKafkaActive() { + return isKafkaActive; + } + + /** + * @param topicName The topic from which messages will be fetched + * @return Either A list of messages from a specific topic, or a specific error response + */ + public Either<Iterable<String>, CambriaErrorResponse> fetchFromTopic(String topicName) { + try { + if(sdcKafkaConsumer == null){ + sdcKafkaConsumer = new SdcKafkaConsumer(deConfiguration); + } + sdcKafkaConsumer.subscribe(topicName); + Iterable<String> messages = sdcKafkaConsumer.poll(topicName); + log.info("Returning messages from topic {}", topicName); + return Either.left(messages); + } catch (KafkaException e) { + BeEcompErrorManager.getInstance().logBeDistributionEngineSystemError("fetchFromTopic", e.getMessage()); + log.error("Failed to fetch from kafka for topic: {}", topicName, e); + CambriaErrorResponse cambriaErrorResponse = + new CambriaErrorResponse(CambriaOperationStatus.INTERNAL_SERVER_ERROR, + HttpStatus.SC_INTERNAL_SERVER_ERROR); + return Either.right(cambriaErrorResponse); + } + } + + /** + * Publish notification message to a given topic and flush + * + * @param topicName The topic to which the message should be published + * @param data The data to publish to the topic specified + * @return CambriaErrorResponse a status response on success or any errors thrown + */ + public CambriaErrorResponse sendNotification(String topicName, INotificationData data) { + CambriaErrorResponse response; + if(sdcKafkaProducer == null){ + sdcKafkaProducer = new SdcKafkaProducer(deConfiguration); + } + try { + String json = gson.toJson(data); + log.info("Before sending to topic {}", topicName); + sdcKafkaProducer.send(json, topicName); + } + catch(KafkaException e){ + BeEcompErrorManager.getInstance().logBeDistributionEngineSystemError("sendNotification", e.getMessage()); + log.error("Failed to send message . Exception {}", e.getMessage()); + + return new CambriaErrorResponse(CambriaOperationStatus.INTERNAL_SERVER_ERROR, 500); + } catch (JsonSyntaxException e) { + BeEcompErrorManager.getInstance().logBeDistributionEngineSystemError("sendNotification", e.getMessage()); + log.error("Failed to convert data to json: {}", data, e); + + return new CambriaErrorResponse(CambriaOperationStatus.INTERNAL_SERVER_ERROR, 500); + } finally { + try { + sdcKafkaProducer.flush(); + response = new CambriaErrorResponse(CambriaOperationStatus.OK, 200); + } catch (KafkaException | IllegalArgumentException e) { + BeEcompErrorManager.getInstance().logBeDistributionEngineSystemError("sendNotification", e.getMessage()); + log.error("Failed to flush sdcKafkaProducer", e); + + response = new CambriaErrorResponse(CambriaOperationStatus.INTERNAL_SERVER_ERROR, 500); + } + } + + return response; + } +} |