fix: make committed offset accurate on partition assignment (confluentinc#893)

Cherry-pick of confluentinc#893 (author: sangreal).

Race condition in PartitionState: createOffsetAndMetadata() called
tryToEncodeOffsets() and getOffsetToCommit() separately. Between the
two calls, incompletes could drain, causing a higher offset to be
committed than intended. After rebalance, the consumer fetches a
non-existent offset and triggers auto.offset.reset (data loss or
replay).

Fix: tryToEncodeOffsets() now calls getOffsetToCommit() once at the
top and returns a Tuple<Optional<String>, Long> so the offset and
payload are computed atomically from the same state snapshot.

Upstream PR: confluentinc#893
Approved by Roman Kolesnev, run in production for >1 week.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: astubbs

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

@@ -4,6 +4,7 @@
  * Copyright (C) 2020-2024 Confluent, Inc.
  */
 
+import io.confluent.parallelconsumer.ParallelConsumer;
 import io.confluent.parallelconsumer.internal.BrokerPollSystem;
 import io.confluent.parallelconsumer.internal.EpochAndRecordsMap;
 import io.confluent.parallelconsumer.internal.PCModule;
@@ -410,8 +411,11 @@ public Optional<OffsetAndMetadata> getCommitDataIfDirty() {
 
     // visible for testing
     protected OffsetAndMetadata createOffsetAndMetadata() {
-        Optional<String> payloadOpt = tryToEncodeOffsets();
-        long nextOffset = getOffsetToCommit();
+        // use tuple to make sure getOffsetToCommit is invoked only once to avoid dirty read
+        // and commit the wrong offset
+        ParallelConsumer.Tuple<Optional<String>, Long> tuple = tryToEncodeOffsets();
+        Optional<String> payloadOpt = tuple.getLeft();
+        long nextOffset = tuple.getRight();
         return payloadOpt
                 .map(encodedOffsets -> new OffsetAndMetadata(nextOffset, encodedOffsets))
                 .orElseGet(() -> new OffsetAndMetadata(nextOffset));
@@ -481,29 +485,30 @@ public long getOffsetHighestSequentialSucceeded() {
      *
      * @return if possible, the String encoded offset map
      */
-    private Optional<String> tryToEncodeOffsets() {
+    private ParallelConsumer.Tuple<Optional<String>, Long> tryToEncodeOffsets() {
+        long offsetOfNextExpectedMessage = getOffsetToCommit();
+
         if (incompleteOffsets.isEmpty()) {
             setAllowedMoreRecords(true);
-            return empty();
+            return ParallelConsumer.Tuple.pairOf(empty(), offsetOfNextExpectedMessage);
         }
 
         try {
             // todo refactor use of null shouldn't be needed. Is OffsetMapCodecManager stateful? remove null #233
-            long offsetOfNextExpectedMessage = getOffsetToCommit();
             var offsetRange = getOffsetHighestSucceeded() - offsetOfNextExpectedMessage;
             String offsetMapPayload = om.makeOffsetMetadataPayload(offsetOfNextExpectedMessage, this);
             ratioPayloadUsedDistributionSummary.record(offsetMapPayload.length() / (double) offsetRange);
             ratioMetadataSpaceUsedDistributionSummary.record(offsetMapPayload.length() / (double) OffsetMapCodecManager.DefaultMaxMetadataSize);
             boolean mustStrip = updateBlockFromEncodingResult(offsetMapPayload);
             if (mustStrip) {
-                return empty();
+                return ParallelConsumer.Tuple.pairOf(empty(), offsetOfNextExpectedMessage);
             } else {
-                return of(offsetMapPayload);
+                return ParallelConsumer.Tuple.pairOf(of(offsetMapPayload), offsetOfNextExpectedMessage);
             }
         } catch (NoEncodingPossibleException e) {
             setAllowedMoreRecords(false);
             log.warn("No encodings could be used to encode the offset map, skipping. Warning: messages might be replayed on rebalance.", e);
-            return empty();
+            return ParallelConsumer.Tuple.pairOf(empty(), offsetOfNextExpectedMessage);
         }
     }