double-check number of requests before pausing the consumer

bsideup · May 7, 2018 · e453515 · e453515
1 parent 658bab9
commit e453515
Showing 1 changed file with 4 additions and 2 deletions.
diff --git a/...a-records-storage/src/main/java/com/github/bsideup/liiklus/kafka/KafkaRecordsStorage.java b/...a-records-storage/src/main/java/com/github/bsideup/liiklus/kafka/KafkaRecordsStorage.java
@@ -122,9 +122,11 @@ public Subscription subscribe(String topic, String groupId, Optional<String> aut
                                             recordFlux
                                                     .filter(it -> it.getPartition() == partitionNum)
                                                     .delayUntil(record -> {
-                                                        if (requests.decrementAndGet() <= 0) {
+                                                        if (requests.decrementAndGet() < 0) {
                                                             return kafkaReceiver.doOnConsumer(consumer -> {
-                                                                consumer.pause(partitionList);
+                                                                if (requests.get() < 0) {
+                                                                    consumer.pause(partitionList);
+                                                                }
                                                                 return true;
                                                             });
                                                         } else {