Would it make sense to have something like (T, Timestamp) => Option[T]? This way you could flatMap kv pairs out of the previous batch instead of just filtering them.

Say you have a T = (K, Map[A,DecayedValue]) and assuming K hasn't been observed in the current batches, you could evict counts that have gone to zero.