get_offset.scala

def lastOffset(consumer: SimpleConsumer, topic: String, partitionId: Int): Future[Long] = {
      val topicAndPartition = new TopicAndPartition(topic, partitionId)
      val requestInfo = Map(
        topicAndPartition -> new PartitionOffsetRequestInfo(OffsetRequest.LatestTime, 1)
      )
      val request = new OffsetRequest(requestInfo, correlationId = Random.nextInt())
      val response = Future(consumer.getOffsetsBefore(request))
      response map { response =>
        response.offsetsGroupedByTopic(topic)(topicAndPartition).offsets.head
      }
    }