fix for message loss for issue 826 (#827)

* use offsetHighestSucceeded instead of offsetHighestSeen to fix issue #826

* update readme

* add ut

* add comments

* remove ut

---------

Co-authored-by: Roman Kolesnev <[email protected]>

Author: sangreal

CHANGELOG.adoc

@@ -18,6 +18,7 @@ endif::[]
 
 === Fixes
 
+* fix: message loss on closing or partitions revoked (#826)
 * fix: ConcurrentModificationException Happened while high load and draining (#822) fixes (#821)
 * fix: safely completing doClose() (#818) partially fixes (#809)
 * Improved offset commit retry. Add support for SaslAuthenticationException retry timeout (#819), partially fixes (#809) in Commit_Sync mode

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/PartitionState.java

@@ -240,7 +240,7 @@ public boolean isRecordPreviouslyCompleted(final ConsumerRecord<K, V> rec) {
             return false;
         } else {
             // if within the range of tracked offsets, must have been previously completed, as it's not in the incomplete set
-            return recOffset <= offsetHighestSeen;
+            return recOffset <= offsetHighestSucceeded;
         }
     }
 
@@ -459,7 +459,8 @@ public long getOffsetHighestSequentialSucceeded() {
          *
          * See #200 for the complete correct solution.
          */
-        long currentOffsetHighestSeen = offsetHighestSeen;
+        // use offsetHighestSucceeded instead of offsetHighestSeen to fix issue #826
+        long currentOffsetHighestSeen = offsetHighestSucceeded;
         Long firstIncompleteOffset = incompleteOffsets.keySet().ceiling(KAFKA_OFFSET_ABSENCE);
         boolean incompleteOffsetsWasEmpty = firstIncompleteOffset == null;