Feature/212 improve logging (#213)

* Improve logging

* Refactoring

Author: kevinwallimann

ingestor-default/src/main/scala/za/co/absa/hyperdrive/ingestor/implementation/transformer/deduplicate/kafka/DeduplicateKafkaSinkTransformer.scala

@@ -17,15 +17,14 @@ package za.co.absa.hyperdrive.ingestor.implementation.transformer.deduplicate.ka
 
 import java.time.Duration
 import java.util.{Properties, UUID}
-
 import za.co.absa.hyperdrive.ingestor.api.utils.ConfigUtils
 import za.co.absa.hyperdrive.ingestor.implementation.transformer.avro.confluent.{ConfluentAvroDecodingTransformer, ConfluentAvroEncodingTransformer}
 import za.co.absa.hyperdrive.ingestor.implementation.utils.KafkaUtil
-
 import org.apache.avro.generic.GenericRecord
 import org.apache.commons.configuration2.Configuration
 import org.apache.hadoop.fs.Path
 import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord, KafkaConsumer}
+import org.apache.kafka.common.TopicPartition
 import org.apache.logging.log4j.LogManager
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.execution.streaming.{CommitLog, OffsetSeqLog}
@@ -74,9 +73,14 @@ private[transformer] class DeduplicateKafkaSinkTransformer(
     implicit val kafkaConsumerTimeoutImpl: Duration = kafkaConsumerTimeout
     val sourceConsumer = createConsumer(readerBrokers, readerExtraOptions, readerSchemaRegistryUrl)
     val latestCommittedOffsets = KafkaUtil.getLatestCommittedOffset(offsetLog, commitLog)
+    logCommittedOffsets(latestCommittedOffsets)
+
     KafkaUtil.seekToOffsetsOrBeginning(sourceConsumer, readerTopic, latestCommittedOffsets)
+    logCurrentPositions(sourceConsumer)
 
     val latestOffsetsOpt = KafkaUtil.getLatestOffset(offsetLog)
+    logOffsets(latestOffsetsOpt)
+
     val sourceRecords = latestOffsetsOpt.map(latestOffset => consumeAndClose(sourceConsumer,
       consumer => KafkaUtil.getMessagesAtLeastToOffset(consumer, latestOffset))).getOrElse(Seq())
     val sourceIds = sourceRecords.map(extractIdFieldsFromRecord(_, sourceIdColumnNames))
@@ -86,14 +90,51 @@ private[transformer] class DeduplicateKafkaSinkTransformer(
     val recordsPerPartition = sinkTopicPartitions.map(p => p -> sourceRecords.size.toLong).toMap
     val latestSinkRecords = consumeAndClose(sinkConsumer, consumer =>
       KafkaUtil.getAtLeastNLatestRecordsFromPartition(consumer, recordsPerPartition))
-    val publishedIds = latestSinkRecords.map(extractIdFieldsFromRecord(_, destinationIdColumnNames))
+    logConsumedSinkRecords(latestSinkRecords)
 
+    val publishedIds = latestSinkRecords.map(extractIdFieldsFromRecord(_, destinationIdColumnNames))
     val duplicatedIds = sourceIds.intersect(publishedIds)
+    logDuplicatedIds(duplicatedIds)
     val duplicatedIdsLit = duplicatedIds.map(duplicatedId => struct(duplicatedId.map(lit): _*))
     val idColumns = sourceIdColumnNames.map(col)
     dataFrame.filter(not(struct(idColumns: _*).isInCollection(duplicatedIdsLit)))
   }
 
+  private def logCommittedOffsets(offsets: Option[Map[TopicPartition, Long]]): Unit =
+    logger.info(s"Latest committed source offsets by partition for ${readerTopic}: { ${offsetsToString(offsets)} }" )
+
+  private def logOffsets(offsets: Option[Map[TopicPartition, Long]]): Unit =
+    logger.info(s"Latest source offsets by partition for ${readerTopic}: { ${offsetsToString(offsets)} }" )
+
+  private def offsetsToString(offsets: Option[Map[TopicPartition, Long]]) = {
+    offsets.map(_.toSeq
+      .sortBy{ case (tp, _) => tp.partition()}
+      .map{ case (tp, offset) => s"${tp.partition()}: $offset"}.reduce(_ + ", " + _)).getOrElse("-")
+  }
+
+  private def logCurrentPositions(consumer: KafkaConsumer[GenericRecord, GenericRecord]): Unit = {
+    val sourcePartitions = KafkaUtil.getTopicPartitions(consumer, readerTopic)
+    val currentPositions = sourcePartitions
+      .sortBy(_.partition())
+      .map { tp => s"${tp.partition()}: ${consumer.position(tp)}"}.reduce(_ + ", " + _)
+    logger.info(s"Reset source offsets by partition to { ${currentPositions} }")
+  }
+
+  private def logConsumedSinkRecords(latestSinkRecords: Seq[ConsumerRecord[GenericRecord, GenericRecord]]): Unit = {
+    val offsetsByPartition = latestSinkRecords.map(r => r.partition() -> r.offset())
+      .groupBy(_._1)
+      .mapValues(_.map(_._2))
+      .toSeq
+      .sortBy(_._1)
+    val firstOffsets = offsetsByPartition.map { case (partition, offsets) => s"$partition: ${offsets.take(3)}"}.reduce(_ + ", " + _)
+    val lastOffsets = offsetsByPartition.map { case (partition, offsets) => s"$partition: ${offsets.takeRight(3)}"}.reduce(_ + ", " + _)
+    logger.info(s"Consumed ${latestSinkRecords.size} sink records. First three offsets by partition: { ${firstOffsets} }. Last three offsets: { ${lastOffsets} }")
+  }
+
+  private def logDuplicatedIds(duplicatedIds: Seq[Seq[Any]]): Unit = {
+    logger.info(s"Found ${duplicatedIds.size} duplicated ids. First three: ${duplicatedIds.take(3)}.")
+  }
+
   private def extractIdFieldsFromRecord(record: ConsumerRecord[GenericRecord, GenericRecord], idColumnNames: Seq[String]): Seq[Any] = {
     idColumnNames.map(idColumnName =>
       AvroUtil.getFromConsumerRecord(record, idColumnName)

ingestor-default/src/main/scala/za/co/absa/hyperdrive/ingestor/implementation/utils/KafkaUtil.scala

@@ -99,7 +99,6 @@ private[hyperdrive] object KafkaUtil {
   private def offsetsHaveBeenReached[K, V](consumer: KafkaConsumer[K, V], toOffsets: Map[TopicPartition, Long]) = {
     toOffsets.forall { case (tp, toOffset) =>
       val position = consumer.position(tp)
-      logger.info(s"Reached position $position on topic partition $tp. Target offset is $toOffset")
       position >= toOffset
     }
   }