AutoCommit polling logic (monix#94)

voidconductor · Sep 11, 2019 · a287fc3 · a287fc3
1 parent 7884fd9
commit a287fc3
Showing 1 changed file with 6 additions and 0 deletions.
diff --git a/kafka-1.0.x/src/main/scala/monix/kafka/KafkaConsumerObservableAutoCommit.scala b/kafka-1.0.x/src/main/scala/monix/kafka/KafkaConsumerObservableAutoCommit.scala
@@ -68,12 +68,16 @@ final class KafkaConsumerObservableAutoCommit[K, V] private[kafka] (
       s.executeAsync { () =>
         val ackFuture =
           try consumer.synchronized {
+            val assignment = consumer.assignment()
             if (cancelable.isCanceled) Stop
             else {
+              consumer.resume(assignment)
               val next = blocking(consumer.poll(pollTimeoutMillis))
+              consumer.pause(assignment)
               if (shouldCommitBefore) consumerCommit(consumer)
               // Feeding the observer happens on the Subscriber's scheduler
               // if any asynchronous boundaries happen
+              isAcked.set(false)
               Observer.feed(out, next.asScala)(out.scheduler)
             }
           } catch {
@@ -83,6 +87,7 @@ final class KafkaConsumerObservableAutoCommit[K, V] private[kafka] (
 
         ackFuture.syncOnComplete {
           case Success(ack) =>
+            isAcked.set(true)
             // The `streamError` flag protects against contract violations
             // (i.e. onSuccess/onError should happen only once).
             // Not really required, but we don't want to depend on the
@@ -106,6 +111,7 @@ final class KafkaConsumerObservableAutoCommit[K, V] private[kafka] (
             }
 
           case Failure(ex) =>
+            isAcked.set(true)
             asyncCb.onError(ex)
         }
       }