   (:import (org.apache.kafka.clients.consumer ConsumerRecords ConsumerRecord)
            (org.apache.kafka.clients.consumer KafkaConsumer ConsumerRebalanceListener Consumer)
            (franzy.clients.consumer.client FranzConsumer)
+           [java.util.concurrent.atomic AtomicLong]
            (org.apache.kafka.clients.producer Callback)
            [franzy.clients.producer.types ProducerRecord]
-           [org.apache.kafka.common TopicPartition]
-           [clojure.core.async.impl.channels ManyToManyChannel]))
+           [org.apache.kafka.common TopicPartition]))
 
 (def defaults
   {:kafka/receive-buffer-bytes 65536

  (recover! [this replica-version checkpoint]
    (set! drained false)
    (set! iter nil)
    (seek-offset! log-prefix consumer kpartition task-map topic checkpoint)
    this)


    (if (and iter (.hasNext ^java.util.Iterator iter))
      (let [rec ^ConsumerRecord (.next ^java.util.Iterator iter)
            new-offset (if rec (.offset rec) offset)
            deserialized (some-> rec segment-fn)]
        (if (= :done deserialized)
          (do (set! drained true)
              nil)
          (do
           (.set ^AtomicLong read-offset new-offset)
           (set! offset new-offset) 
           deserialized))) 
      (do (set! iter 
