<commit_before>package com.ubirch.avatar.core.kafka

import akka.actor.ActorSystem
import akka.kafka.ProducerSettings
import com.typesafe.scalalogging.StrictLogging
import com.ubirch.avatar.config.Config
import org.apache.kafka.clients.producer.{Callback, Producer, ProducerRecord, RecordMetadata}
import org.apache.kafka.common.serialization.StringSerializer

import scala.concurrent.{ExecutionContext, Future, Promise}
import scala.util.control.NonFatal

class KafkaProducer(kafkaUrl: String, topic: String, actorSystem: ActorSystem) {

  val producerSettings: ProducerSettings[String, String] =
    if (Config.isSecureKafkaConnection) {
      ProducerSettings(actorSystem, new StringSerializer, new StringSerializer)
        .withBootstrapServers(kafkaUrl)
        .withProperties(Config.kafkaProdSecureConnectionProperties)
    } else {
      ProducerSettings(actorSystem, new StringSerializer, new StringSerializer)
        .withBootstrapServers(kafkaUrl)
    }

  val producer: Producer[String, String] = producerSettings.createKafkaProducer()

  def send(message: String)(implicit ec: ExecutionContext): Future[RecordMetadata] = {
    val record = new ProducerRecord[String, String](topic, message)
    val promise = Promise[RecordMetadata]()
    try {
      producer.send(record, new Callback {
          override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit = {
            if(exception == null) promise.success(metadata)
            else promise.failure(exception)
          }
        })
    } catch {
      case NonFatal(e) => promise.failure(e)
    }
    promise.future
  }

  def close(): Unit = {
    producer.close()
  }
}

object KafkaProducer extends StrictLogging {
  /**
   * throw exception when the initialization is failed
   */
  def create(kafkaUrl: String, topic: String, actorSystem: ActorSystem): KafkaProducer =
    try {
      logger.info(s"Trying to initialize kafka producer connection. IsSecureConnection: ${Config.isSecureKafkaConnection}")
      new KafkaProducer(kafkaUrl, topic, actorSystem)
    }
    catch { case ex: Exception =>
      logger.error(s"Error occurred while initializing Kafka producer! " +
        s"Message: ${ex.getMessage} Kafka URL: $kafkaUrl Properties: ${Config.kafkaProdSecureConnectionProperties}")
      throw ex
    }

}<commit_msg>UC-669 Better logging<commit_after>package com.ubirch.avatar.core.kafka

import akka.actor.ActorSystem
import akka.kafka.ProducerSettings
import com.typesafe.scalalogging.StrictLogging
import com.ubirch.avatar.config.Config
import org.apache.kafka.clients.producer.{Callback, Producer, ProducerRecord, RecordMetadata}
import org.apache.kafka.common.serialization.StringSerializer

import scala.concurrent.{ExecutionContext, Future, Promise}
import scala.util.control.NonFatal

class KafkaProducer(kafkaUrl: String, topic: String, actorSystem: ActorSystem) {

  val producerSettings: ProducerSettings[String, String] =
    if (Config.isSecureKafkaConnection) {
      ProducerSettings(actorSystem, new StringSerializer, new StringSerializer)
        .withBootstrapServers(kafkaUrl)
        .withProperties(Config.kafkaProdSecureConnectionProperties)
    } else {
      ProducerSettings(actorSystem, new StringSerializer, new StringSerializer)
        .withBootstrapServers(kafkaUrl)
    }

  val producer: Producer[String, String] = producerSettings.createKafkaProducer()

  def send(message: String)(implicit ec: ExecutionContext): Future[RecordMetadata] = {
    val record = new ProducerRecord[String, String](topic, message)
    val promise = Promise[RecordMetadata]()
    try {
      producer.send(record, new Callback {
          override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit = {
            if(exception == null) promise.success(metadata)
            else promise.failure(exception)
          }
        })
    } catch {
      case NonFatal(e) => promise.failure(e)
    }
    promise.future
  }

  def close(): Unit = {
    producer.close()
  }
}

object KafkaProducer extends StrictLogging {
  /**
   * throw exception when the initialization is failed
   */
  def create(kafkaUrl: String, topic: String, actorSystem: ActorSystem): KafkaProducer =
    try {
      logger.info(s"Trying to initialize kafka producer connection." +
        s"IsSecureConnection: ${Config.isSecureKafkaConnection} Kafka URL: $kafkaUrl")
      new KafkaProducer(kafkaUrl, topic, actorSystem)
    }
    catch { case ex: Exception =>
      logger.error(s"Error occurred while initializing Kafka producer! ${ex.getMessage}")
      throw ex
    }

}
{'fraction_non_alphanumeric': 0.09450021824530773, 'fraction_numerical': 0.0006547359231776517, 'mean_word_length': 4.052921719955899, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3110053', 'n_tokens_mistral': 1315, 'n_tokens_neox': 1307, 'n_words': 291}