Faust commits the wrong offset in case of a gap in acks #312 (#313)

* Faust commits the wrong offset in case of a gap in acks #312

Faust commits the wrong offset in case of a gap in acks #312

Co-Authored-By: ekerstens <49325583+ekerstens@users.noreply.github.com>

* kljlk

jhkjhj

* Revert "kljlk"

This reverts commit 8b487fc.

* update fix #313

Co-Authored-By: ekerstens <49325583+ekerstens@users.noreply.github.com>

* update test case

Co-Authored-By: ekerstens <49325583+ekerstens@users.noreply.github.com>
Co-Authored-By: Leo <47164895+LeohuangLeo@users.noreply.github.com>

Co-authored-by: Leo Huang <s3563602123@gmail.com>
Co-authored-by: ekerstens <49325583+ekerstens@users.noreply.github.com>
Co-authored-by: Leo <47164895+LeohuangLeo@users.noreply.github.com>

Author: 3 people

faust/transport/consumer.py

@@ -1097,6 +1097,18 @@ def _new_offset(self, tp: TP) -> Optional[int]:
                 acked.extend(gaps)
                 gap_for_tp[:gap_index] = []
             acked.sort()
+
+            # We iterate over it until we handle gap in the head of acked queue
+            # then return the previous committed offset.
+            # For example if acked[tp] is:
+            #    34 35 36 37
+            #  ^-- gap
+            # self._committed_offset[tp] is 31
+            # the return value will be None (the same as 31)
+            if self._committed_offset[tp]:
+                if min(acked) - self._committed_offset[tp] > 0:
+                    return None
+
             # Note: acked is always kept sorted.
             # find first list of consecutive numbers
             batch = next(consecutive_numbers(acked))

tests/unit/transport/test_consumer.py

@@ -1079,9 +1079,13 @@ def test_new_offset(self, tp, acked, expected_offset, expected_acked, *, consume
             (TP1, [1, 2, 3, 4, 5, 6, 7, 8, 10], [9], 11),
             (TP1, [1, 2, 3, 4, 6, 7, 8, 10], [5], 9),
             (TP1, [1, 3, 4, 6, 7, 8, 10], [2, 5, 9], 11),
+            (TP1, [3, 4], [], None),
+            (TP1, [3, 4], [2], None),
+            (TP1, [3, 4], [1, 2], 5),
         ],
     )
     def test_new_offset_with_gaps(self, tp, acked, gaps, expected_offset, *, consumer):
+        consumer._committed_offset[tp] = 1
         consumer._acked[tp] = acked
         consumer._gap[tp] = gaps
         assert consumer._new_offset(tp) == expected_offset