KAFKA-19781: Fix to not update positions for partitions being revoked (#20914)

Fix to avoid initializing positions for partitions being revoked, as it
is unneeded (we do not allow fetching from partitions being revoked),
and could lead to NoOffsetForPartitionException on a partition that is
already being revoked (this is confusing for applications).

This is the behaviour we already had for fetch, just applying it to
update positions to align.

This gap was surfaced on edge cases of partitions being assigned and
revoked right away.

Reviewers: Andrew Schofield <[email protected]>, Lucas Brutschy
 <[email protected]>

Author: lianetm

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

@@ -1202,14 +1202,16 @@ private void resume() {
         }
 
         /**
-         * True if the partition is in {@link FetchStates#INITIALIZING} state. While in this
-         * state, a position for the partition can be retrieved (based on committed offsets or
+         * Check if we need to retrieve a fetch position for the given partition.
+         * True if the partition state is {@link FetchStates#INITIALIZING}, and the partition is not being revoked.
+         * <p/>
+         * While in this state, a position for the partition will be retrieved (based on committed offsets or
          * partitions offsets).
          * Note that retrieving a position does not mean that we can start fetching from the
          * partition (see {@link #isFetchable()})
          */
         private boolean shouldInitialize() {
-            return fetchState.equals(FetchStates.INITIALIZING);
+            return fetchState.equals(FetchStates.INITIALIZING) && !pendingRevocation;
         }
 
         private boolean isFetchable() {

clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java

@@ -306,15 +306,24 @@ public void partitionPause() {
     }
 
     @Test
-    public void testMarkingPartitionPending() {
+    public void testMarkingPendingRevocation() {
         state.assignFromUser(Set.of(tp0));
         state.seek(tp0, 100);
         assertTrue(state.isFetchable(tp0));
+        assertFalse(state.isPaused(tp0));
         state.markPendingRevocation(Set.of(tp0));
         assertFalse(state.isFetchable(tp0));
         assertFalse(state.isPaused(tp0));
     }
 
+    @Test
+    public void testMarkingPendingRevocationPreventsInitializingPosition() {
+        state.assignFromUser(Set.of(tp0));
+        assertTrue(state.initializingPartitions().contains(tp0));
+        state.markPendingRevocation(Set.of(tp0));
+        assertFalse(state.initializingPartitions().contains(tp0));
+    }
+
     @Test
     public void testAssignedPartitionsAwaitingCallbackKeepPositionDefinedInCallback() {
         // New partition assigned. Should not be fetchable or initializing positions.