@johnynek I have never used this before but here is the first stab at it. Were you thinking something along this line? Let me know what else you want me to change
```
override def multiPut[K1 <: K](kvs: Map[K1, V]): Map[K1, Future[Unit]] = {
    val s = new AsyncSemaphore(permits)
    val future = s.acquire()
    future.onSuccess {
      p =>
        val batch = kvs.map {
          case (k, v) => new ProducerData[K, V](topic, k, List(v))
        }.toList
        producer.send(batch: _*)
    }
    kvs.mapValues(v => future.unit)
  }
}

```