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:
us...@infra.apache.org
