KafkaCluster.scala

def getCommittedOffsetsForConsumer(consumerConfig: KafkaConsumerConfig, topicPartitions: Seq[TopicPartition]): Map[TopicPartition, Long] = {
    val result = withCustomStringConsumer(consumer => {
      consumer.assign(topicPartitions.toList.asJava)
      val tps = topicPartitions.map( tp => {
        val offsetAndMetadata = consumer.committed(tp)
        val startOffset: Long = KafkaCluster.getLogBeginningOffset(consumerConfig.topic).values.head
        
        Option(offsetAndMetadata) match {
          case None => {
            // log "TopicPartition $tp doesn't have committed offsets"
            tp -> startOffset
          }
          case Some(om) => {
            val currentOffsets = om.offset()
            if (currentOffsets < startOffset) {
              tp -> startOffset
            } else {
              tp -> currentOffsets
            }
          }
        }
      }
      ).toMap

      consumer.close()

      tps
    })(consumerConfig)


      result.asInstanceOf[Map[TopicPartition, Long]]
  }