Minor fixes to InfluxDB and Kafka sinks

Author: LucaCanali

src/main/scala/ch/cern/sparkmeasure/InfluxDBSink.scala

@@ -1,6 +1,6 @@
 package ch.cern.sparkmeasure
 
-import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd, SparkListenerApplicationStart, SparkListenerEvent, SparkListenerExecutorAdded, SparkListenerJobEnd, SparkListenerJobStart, SparkListenerStageCompleted, SparkListenerStageSubmitted, SparkListenerTaskEnd, SparkListenerTaskStart, TaskLocality}
+import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd, SparkListenerApplicationStart, SparkListenerEvent, SparkListenerExecutorAdded, SparkListenerJobEnd, SparkListenerJobStart, SparkListenerStageCompleted, SparkListenerStageSubmitted, SparkListenerTaskEnd, SparkListenerTaskStart}
 import org.apache.spark.sql.execution.ui.{SparkListenerSQLExecutionEnd, SparkListenerSQLExecutionStart}
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.SparkConf

src/main/scala/ch/cern/sparkmeasure/KafkaSink.scala

@@ -221,15 +221,19 @@ class KafkaSink(conf: SparkConf) extends SparkListener {
     )
   }
 
-
-  protected def report[T <: Any](metrics: Map[String, T]): Unit = Try {
-    ensureProducer()
-
-    val str = IOUtils.writeToStringSerializedJSON(metrics)
-    val message = str.getBytes(StandardCharsets.UTF_8)
-    producer.send(new ProducerRecord[String, Array[Byte]](topic, message))
-  }.recover {
-    case ex: Throwable => logger.error(s"error on reporting metrics to kafka stream, details=${ex.getMessage}", ex)
+  protected def report[T <: Any](metrics: Map[String, T]): Unit = {
+    val result: Unit = Try {
+      ensureProducer()
+      val str = IOUtils.writeToStringSerializedJSON(metrics)
+      val message = str.getBytes(StandardCharsets.UTF_8)
+      producer.send(new ProducerRecord[String, Array[Byte]](topic, message))
+    }.recover {
+      case ex: Throwable =>
+        logger.error(s"Error on reporting metrics to Kafka stream, details=${ex.getMessage}", ex)
+        () // Explicitly return Unit
+    }.getOrElse(()) // Ensure Unit is the final type
+
+    result // Return Unit
   }
 
   private def ensureProducer(): Unit = {