Skip commit when records offset < initial offset (knative-extensions#4348) (#1756)

* Skip recordNewOffset when records offset < initial offset

* Run hack/update-codegen.sh

* Make recordReceived a bit safer concurrency-wise

* Fix: Remove redundant if clause

Co-authored-by: Christoph Stäbler <[email protected]>

Author: openshift-cherrypick-robot

data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java

@@ -102,9 +102,12 @@ public PartitionRevokedHandler getPartitionRevokedHandler() {
     @Override
     public void recordReceived(final ConsumerRecord<?, ?> record) {
         final var tp = new TopicPartition(record.topic(), record.partition());
-        if (!offsetTrackers.containsKey(tp)) {
-            // Initialize offset tracker for the given record's topic/partition.
-            offsetTrackers.putIfAbsent(tp, new OffsetTracker(record.offset()));
+        final var offsetTracker = offsetTrackers.putIfAbsent(tp, new OffsetTracker(record.offset()));
+        if (offsetTracker != null && record.offset() < offsetTracker.initialOffset) {
+            logger.debug(
+                    "Received records offset ({}) is less than offsetTrackers initial offset ({})",
+                    record.offset(),
+                    offsetTracker.initialOffset);
         }
     }
 
@@ -148,6 +151,16 @@ private void commit(final ConsumerRecord<?, ?> record) {
         // Note: it's not possible to commit offsets of partitions that this a particular consumer instance doesn't own.
         final var ot = this.offsetTrackers.get(new TopicPartition(record.topic(), record.partition()));
         if (ot != null) {
+            if (record.offset() < ot.initialOffset) {
+                logger.debug(
+                        "Ignoring commit for {}-{} offset {}, offset tracker already on {}",
+                        record.topic(),
+                        record.partition(),
+                        record.offset(),
+                        ot.initialOffset);
+                return;
+            }
+
             ot.recordNewOffset(record.offset());
         }
     }