KafkaConsumer.seekToEnd cannot take effect when integrating with spark streaming

protected def latestOffsets(): Map[TopicPartition, Long] = {
    val c = consumer
    c.poll(0)
    val parts = c.assignment().asScala
    val newPartitions = parts.diff(currentOffsets.keySet)
    currentOffsets = currentOffsets ++ newPartitions.map(tp => tp -> c.position(tp)).toMap
    c.pause(newPartitions.asJava)
    c.seekToEnd(currentOffsets.keySet.asJava)
    parts.map(tp => tp -> c.position(tp)).toMap
  }

        fetcher.resetOffsetsIfNeeded(partitions);    // reset to latest offset for current partition
        if (!subscriptions.hasAllFetchPositions()) {  // called seekToEnd for all partitions before, so this sentence will be true 
            coordinator.refreshCommittedOffsetsIfNeeded();
            fetcher.updateFetchPositions(partitions);  // reset to committed offsets for current partition
        }