Scala clock cache

ClockCache.scala

import com.twitter.algebird.Semigroup


case class ClockCache[K, V](cacheSize: Int, maxHits: Int = 10 )
              (implicit sg: Semigroup[V]) {
  private class CacheElement {
    var hits: Int = 0
    var cacheHit: Int = 0
    var key: Any = null
    var value: Any = null
    def reset {
        hits = 0
        key = null
        value = null
        cacheHit =0
    }
  }
  private var position: Int = 0
  private val cache = Array.fill(cacheSize)(new CacheElement)

  def keyUpdate(kv: (K, V), cur: CacheElement): Option[(K, V)] = {
    val newV = sg.plus(cur.value.asInstanceOf[V], kv._2)
    if(cur.hits >= maxHits) {
        cur.reset
        Some((kv._1.asInstanceOf[K], newV.asInstanceOf[V]))
    } else {
      cur.hits += 1
      cur.cacheHit = 1
      cur.value = newV
      None
    }
  }

  def maybeOverwrite(kv: (K, V), cur: CacheElement): Option[(K, V)] = {
    val ret = if(cur.key == null) {
        None
    } else {
        val r = Some((cur.key.asInstanceOf[K], cur.value.asInstanceOf[V]))
        cur.reset
        r
    }
    cur.hits = 1
    cur.key = kv._1
    cur.value = kv._2
    cur.cacheHit = 1
    ret
  }

  def insert(kv: (K, V)): Option[(K, V)] = {
    while(true) {
      val cur = cache(position)
      if(cur.key == kv._1) {
        return keyUpdate(kv, cur)
      } else if(cur.cacheHit == 0) {
        return maybeOverwrite(kv, cur)
      } else {
          cur.cacheHit = 0
      }
      position = (position + 1) % cacheSize
    }
    sys.error("Should never reach here")
  }
}