src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/KafkaPublisher.java

   1 /*******************************************************************************
   2  *  ============LICENSE_START=======================================================
   3  *  org.onap.dmaap
   4  *  ================================================================================
   5  *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
   6  *  ================================================================================
   7  *  Licensed under the Apache License, Version 2.0 (the "License");
   8  *  you may not use this file except in compliance with the License.
   9  *  You may obtain a copy of the License at
  10  *        http://www.apache.org/licenses/LICENSE-2.0
  11 *
  12  *  Unless required by applicable law or agreed to in writing, software
  13  *  distributed under the License is distributed on an "AS IS" BASIS,
  14  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  *  See the License for the specific language governing permissions and
  16  *  limitations under the License.
  17  *  ============LICENSE_END=========================================================
  18  *
  19  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  20  *
  21  *******************************************************************************/
  22 package org.onap.dmaap.dmf.mr.backends.kafka;
  23
  24 import java.io.IOException;
  25 import java.util.ArrayList;
  26 import java.util.LinkedList;
  27 import java.util.List;
  28 import java.util.Properties;
  29 import org.apache.kafka.clients.producer.KafkaProducer;
  30 import org.apache.kafka.clients.producer.Producer;
  31 import org.apache.kafka.clients.producer.ProducerRecord;
  32 import org.json.JSONException;
  33 import org.onap.dmaap.dmf.mr.backends.Publisher;
  34 import org.onap.dmaap.dmf.mr.constants.CambriaConstants;
  35 import org.onap.dmaap.dmf.mr.utils.Utils;
  36 import org.springframework.beans.factory.annotation.Qualifier;
  37 import org.springframework.util.StringUtils;
  38 import com.att.eelf.configuration.EELFLogger;
  39 import com.att.eelf.configuration.EELFManager;
  40 import com.att.nsa.drumlin.till.nv.rrNvReadable;
  41
  42
  43
  44 /**
  45  * Sends raw JSON objects into Kafka.
  46  *
  47  * Could improve space: BSON rather than JSON?
  48  *
  49  * @author peter
  50  *
  51  */
  52
  53 public class KafkaPublisher implements Publisher {
  54         /**
  55          * constructor initializing
  56          *
  57          * @param settings
  58          * @throws rrNvReadable.missingReqdSetting
  59          */
  60         public KafkaPublisher(@Qualifier("propertyReader") rrNvReadable settings) throws rrNvReadable.missingReqdSetting {
  61
  62                 final Properties props = new Properties();
  63                 String kafkaConnUrl= com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,"kafka.metadata.broker.list");
  64                 if(StringUtils.isEmpty(kafkaConnUrl)){
  65
  66                         kafkaConnUrl="localhost:9092";
  67                 }
  68
  69
  70             if(Utils.isCadiEnabled()){
  71                 transferSetting( props, "sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='"+Utils.getKafkaproperty()+"';");
  72                 transferSetting( props, "security.protocol", "SASL_PLAINTEXT");
  73                 transferSetting( props, "sasl.mechanism", "PLAIN");
  74             }
  75                 transferSetting( props, "bootstrap.servers",kafkaConnUrl);
  76
  77                 transferSetting( props, "request.required.acks", "1");
  78                 transferSetting( props, "message.send.max.retries", "5");
  79                 transferSetting(props, "retry.backoff.ms", "150");
  80
  81
  82
  83                 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  84                 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  85
  86
  87
  88                 fProducer = new KafkaProducer<>(props);
  89         }
  90
  91         /**
  92          * Send a message with a given topic and key.
  93          *
  94          * @param msg
  95          * @throws FailedToSendMessageException
  96          * @throws JSONException
  97          */
  98         @Override
  99         public void sendMessage(String topic, message msg) throws IOException{
 100                 final List<message> msgs = new LinkedList<>();
 101                 msgs.add(msg);
 102                 sendMessages(topic, msgs);
 103         }
 104
 105         /**
 106          * method publishing batch messages
 107         * This method is commented from 0.8 to 0.11 upgrade
 108          * @param topic
 109          * @param kms
 110          * throws IOException
 111          *
 112         public void sendBatchMessage(String topic, ArrayList<KeyedMessage<String, String>> kms) throws IOException {
 113                 try {
 114                         fProducer.send(kms);
 115
 116                 } catch (FailedToSendMessageException excp) {
 117                         log.error("Failed to send message(s) to topic [" + topic + "].", excp);
 118                         throw new FailedToSendMessageException(excp.getMessage(), excp);
 119                 }
 120
 121         } */
 122
 123
 124         /*
 125          * Kafka 11.0 Interface
 126          * @see com.att.nsa.cambria.backends.Publisher#sendBatchMessageNew(java.lang.String, java.util.ArrayList)
 127          */
 128         public void sendBatchMessageNew(String topic, ArrayList <ProducerRecord<String,String>> kms) throws IOException {
 129                 try {
 130                         for (ProducerRecord<String,String> km : kms) {
 131                                 fProducer.send(km);
 132                         }
 133
 134                 } catch (Exception excp) {
 135                         log.error("Failed to send message(s) to topic [" + topic + "].", excp);
 136                         throw new IOException(excp.getMessage(), excp);
 137                 }
 138
 139         }
 140
 141         /**
 142          * Send a set of messages. Each must have a "key" string value.
 143          *
 144          * @param topic
 145          * @param msg
 146          * @throws FailedToSendMessageException
 147          * @throws JSONException
 148          *
 149         @Override
 150         public void sendMessages(String topic, List<? extends message> msgs)
 151                         throws IOException, FailedToSendMessageException {
 152                 log.info("sending " + msgs.size() + " events to [" + topic + "]");
 153
 154                 final List<KeyedMessage<String, String>> kms = new ArrayList<KeyedMessage<String, String>>(msgs.size());
 155                 for (message o : msgs) {
 156                         final KeyedMessage<String, String> data = new KeyedMessage<String, String>(topic, o.getKey(), o.toString());
 157                         kms.add(data);
 158                 }
 159                 try {
 160                         fProducer.send(kms);
 161
 162                 } catch (FailedToSendMessageException excp) {
 163                         log.error("Failed to send message(s) to topic [" + topic + "].", excp);
 164                         throw new FailedToSendMessageException(excp.getMessage(), excp);
 165                 }
 166         } */
 167         @Override
 168     public void sendMessagesNew(String topic, List<? extends message> msgs) throws IOException {
 169         log.info("sending " + msgs.size() + " events to [" + topic + "]");
 170         try {
 171             for (message o : msgs) {
 172                 final ProducerRecord<String, String> data =
 173                         new ProducerRecord<>(topic, o.getKey(), o.toString());
 174                 fProducer.send(data);
 175             }
 176         } catch (Exception e) {
 177             log.error("Failed to send message(s) to topic [" + topic + "].", e);
 178         }
 179     }
 180
 181
 182         private Producer<String, String> fProducer;
 183
 184   /**
 185    * It sets the key value pair
 186    * @param topic
 187    * @param msg
 188    * @param key
 189    * @param defVal
 190    */
 191         private void transferSetting(Properties props, String key, String defVal) {
 192                 String kafkaProp= com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,"kafka." + key);
 193                 if (StringUtils.isEmpty(kafkaProp)) kafkaProp=defVal;
 194                 props.put(key, kafkaProp);
 195         }
 196
 197         private static final EELFLogger log = EELFManager.getInstance().getLogger(KafkaPublisher.class);
 198
 199         @Override
 200         public void sendMessages(String topic, List<? extends message> msgs) throws IOException {
 201                 // TODO Auto-generated method stub
 202
 203         }
 204 }