[GitHub] [kafka] hachikuji commented on a change in pull request #8841: KAFKA-10123 Fix incorrect value for AWAIT_RESET#hasPosition

[GitBox <gi...@apache.org>]

hachikuji commented on a change in pull request #8841:
URL: https://github.com/apache/kafka/pull/8841#discussion_r438362895



##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
##########
@@ -978,7 +1000,7 @@ public boolean hasValidPosition() {
 
             @Override
             public boolean hasPosition() {

Review comment:
       Would it make sense to move `hasPosition` to `TopicPartitionState`? Then we could just turn this into a null check on `position`.

##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
##########
@@ -675,36 +676,41 @@ private ListOffsetResult fetchOffsetsByTimes(Map<TopicPartition, Long> timestamp
                     completedFetch.partition);
         } else {
             FetchPosition position = subscriptions.position(completedFetch.partition);
-            if (completedFetch.nextFetchOffset == position.offset) {
-                List<ConsumerRecord<K, V>> partRecords = completedFetch.fetchRecords(maxRecords);
-
-                log.trace("Returning {} fetched records at offset {} for assigned partition {}",
-                        partRecords.size(), position, completedFetch.partition);
-
-                if (completedFetch.nextFetchOffset > position.offset) {
-                    FetchPosition nextPosition = new FetchPosition(
-                            completedFetch.nextFetchOffset,
-                            completedFetch.lastEpoch,
-                            position.currentLeader);
-                    log.trace("Update fetching position to {} for partition {}", nextPosition, completedFetch.partition);
-                    subscriptions.position(completedFetch.partition, nextPosition);
-                }
+            if (position != null) {

Review comment:
       I think the invariant that we try to maintain is that we should have a position if we are in the FETCHING state. I'd suggest we detect this in `transitionState` and raise the exception at that point. Otherwise, we could reach an illegal state and the consumer would just stop fetching the partition. Failing fast is probably preferable. What I have in mind is just something like this:
   ```java
           private void transitionState(FetchState newState, Runnable runIfTransitioned) {
               FetchState nextState = this.fetchState.transitionTo(newState);
               if (nextState.equals(newState)) {
                   if (position == null && (nextState == FETCHING || nextState == AWAIT_VALIDATION))
                      throw new IllegalStateException();
                   this.fetchState = nextState;
                   runIfTransitioned.run();
               }
           }
   ```
   




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org