PoAn Yang created KAFKA-19067:
---------------------------------
Summary: AsyncKafkaConsumer may return stale fetch result after
seek operation
Key: KAFKA-19067
URL: https://issues.apache.org/jira/browse/KAFKA-19067
Project: Kafka
Issue Type: Bug
Reporter: PoAn Yang
Assignee: PoAn Yang
The KafkaConsumer sends FetchRequest after it subscribes topics. The
FetchResponse data stores to FetchBuffer. For KafkaConsumer#seek operation, the
FetchState changes to AWAIT_RESET and the consumer sends LIST_OFFSET request.
The state changes back to FETCHING after the consumer receives LIST_OFFSET
response.
If a KafkaConsumer subscribes topics and calls seek function, there may have
stale FetchResponse data in FetchBuffer. For ClassicKafkaConsumer#poll, it gets
data from FetchBuffer first and then calls ConsumerNetworkClient#poll. If there
is stale data in FetchBuffer, the data is ignored because the FetchState is in
AWAIT_RESET. The FetchState in ClassicKafkaConsumer changes back to FETCHING
after ConsumerNetworkClient#poll receives LIST_OFFSET response.
However, for AsyncKafkaConsumer, it may return stale FetchResponse data to
users, because the ConsumerNetworkThread runs in another thread. The FetchState
may changes back to FETCHING before AsyncKafkaConsumer#poll does valid position
check.
Following logs show the case for ClassicKafkaConsumer:
{noformat}
[Consumer clientId=consumer-group-1, groupId=group] Added read_uncommitted
fetch request for partition topic-0 at position FetchPosition{offset=0,
offsetEpoch=Optional.empty,
currentLeader=LeaderAndEpoch{leader=Optional[localhost:50923 (id: 1 rack: null
isFenced: false)], epoch=0}} to node localhost:50923 (id: 1 rack: null
isFenced: false) (org.apache.kafka.clients.consumer.internals.AbstractFetch:471)
[Consumer clientId=consumer-group-1, groupId=group] Sending FETCH request with
header RequestHeader(apiKey=FETCH, apiVersion=17, clientId=consumer-group-1,
correlationId=12, headerVersion=2) and timeout 30000 to node 1:
FetchRequestData(clusterId=null, replicaId=-1,
replicaState=ReplicaState(replicaId=-1, replicaEpoch=-1), maxWaitMs=500,
minBytes=1, maxBytes=52428800, isolationLevel=0, sessionId=0, sessionEpoch=0,
topics=[FetchTopic(topic='topic', topicId=BatA1H3WQ6KdwhZpMq6fOw,
partitions=[FetchPartition(partition=0, currentLeaderEpoch=0, fetchOffset=0,
lastFetchedEpoch=-1, logStartOffset=-1, partitionMaxBytes=1048576,
replicaDirectoryId=AAAAAAAAAAAAAAAAAAAAAA)])], forgottenTopicsData=[],
rackId='') (org.apache.kafka.clients.NetworkClient:604)
### consumer calls seekToBeginning
[Consumer clientId=consumer-group-1, groupId=group] Sending LIST_OFFSETS
request with header RequestHeader(apiKey=LIST_OFFSETS, apiVersion=10,
clientId=consumer-group-1, correlationId=13, headerVersion=2) and timeout 30000
to node 1: ListOffsetsRequestData(replicaId=-1, isolationLevel=0,
topics=[ListOffsetsTopic(name='topic',
partitions=[ListOffsetsPartition(partitionIndex=0, currentLeaderEpoch=0,
timestamp=-2)])], timeoutMs=30000) (org.apache.kafka.clients.NetworkClient:604)
[Consumer clientId=consumer-group-1, groupId=group] Fetch read_uncommitted at
offset 0 for partition topic-0 returned fetch data
PartitionData(partitionIndex=0, errorCode=0, highWatermark=10,
lastStableOffset=10, logStartOffset=0, divergingEpoch=EpochEndOffset(epoch=-1,
endOffset=-1), currentLeader=LeaderIdAndEpoch(leaderId=-1, leaderEpoch=-1),
snapshotId=SnapshotId(endOffset=-1, epoch=-1), abortedTransactions=null,
preferredReadReplica=-1, records=MemoryRecords(size=151,
buffer=java.nio.HeapByteBuffer[pos=0 lim=151 cap=154]))
(org.apache.kafka.clients.consumer.internals.AbstractFetch:203) <-- The
response of FetchRequest which is sent before calling seekToBeginning. The
FetchState is AWAIT_RESET, so the data is ignored.
[Consumer clientId=consumer-group-1, groupId=group] Ignoring fetched records
for partition topic-0 since it no longer has valid position
(org.apache.kafka.clients.consumer.internals.FetchCollector:226)
[Consumer clientId=consumer-group-1, groupId=group] Resetting offset for
partition topic-0 to position FetchPosition{offset=0,
offsetEpoch=Optional.empty,
currentLeader=LeaderAndEpoch{leader=Optional[localhost:50923 (id: 1 rack: null
isFenced: false)], epoch=0}}.
(org.apache.kafka.clients.consumer.internals.SubscriptionState:451) <-- The
result of seekToBeginning.
[Consumer clientId=consumer-group-1, groupId=group] Added read_uncommitted
fetch request for partition topic-0 at position FetchPosition{offset=0,
offsetEpoch=Optional.empty,
currentLeader=LeaderAndEpoch{leader=Optional[localhost:50923 (id: 1 rack: null
isFenced: false)], epoch=0}} to node localhost:50923 (id: 1 rack: null
isFenced: false)
(org.apache.kafka.clients.consumer.internals.AbstractFetch:471) <-- Send
another FetchRequest starts from offset 0.
[Consumer clientId=consumer-group-1, groupId=group] Sending FETCH request with
header RequestHeader(apiKey=FETCH, apiVersion=17, clientId=consumer-group-1,
correlationId=14, headerVersion=2) and timeout 30000 to node 1:
FetchRequestData(clusterId=null, replicaId=-1,
replicaState=ReplicaState(replicaId=-1, replicaEpoch=-1), maxWaitMs=500,
minBytes=1, maxBytes=52428800, isolationLevel=0, sessionId=166492720,
sessionEpoch=1, topics=[], forgottenTopicsData=[], rackId='')
(org.apache.kafka.clients.NetworkClient:604)
{noformat}
Following logs show the case for AsyncKafkaConsumer:
{noformat}
[Consumer clientId=consumer-group-2, groupId=group] Added read_uncommitted
fetch request for partition topic-0 at position FetchPosition{offset=0,
offsetEpoch=Optional.empty,
currentLeader=LeaderAndEpoch{leader=Optional[localhost:50970 (id: 2 rack: null
isFenced: false)], epoch=0}} to node localhost:50970 (id: 2 rack: null
isFenced: false) (org.apache.kafka.clients.consumer.internals.AbstractFetch:471)
[Consumer clientId=consumer-group-2, groupId=group] Sending FETCH request with
header RequestHeader(apiKey=FETCH, apiVersion=17, clientId=consumer-group-2,
correlationId=30, headerVersion=2) and timeout 30000 to node 2:
FetchRequestData(clusterId=null, replicaId=-1,
replicaState=ReplicaState(replicaId=-1, replicaEpoch=-1), maxWaitMs=500,
minBytes=1, maxBytes=52428800, isolationLevel=0, sessionId=0, sessionEpoch=0,
topics=[FetchTopic(topic='topic', topicId=E2BqIjY8RU2mbcUClbcx3A,
partitions=[FetchPartition(partition=0, currentLeaderEpoch=0, fetchOffset=0,
lastFetchedEpoch=-1, logStartOffset=-1, partitionMaxBytes=1048576,
replicaDirectoryId=AAAAAAAAAAAAAAAAAAAAAA)])], forgottenTopicsData=[],
rackId='') (org.apache.kafka.clients.NetworkClient:604)
[Consumer clientId=consumer-group-2, groupId=group] Fetch read_uncommitted at
offset 0 for partition topic-0 returned fetch data
PartitionData(partitionIndex=0, errorCode=0, highWatermark=10,
lastStableOffset=10, logStartOffset=0, divergingEpoch=EpochEndOffset(epoch=-1,
endOffset=-1), currentLeader=LeaderIdAndEpoch(leaderId=-1, leaderEpoch=-1),
snapshotId=SnapshotId(endOffset=-1, epoch=-1), abortedTransactions=null,
preferredReadReplica=-1, records=MemoryRecords(size=151,
buffer=java.nio.HeapByteBuffer[pos=0 lim=151 cap=154]))
(org.apache.kafka.clients.consumer.internals.AbstractFetch:203) <-- The data of
FetchRequest before calling seekToBeginning, but it has not been used by
application thread.
### consumer calls seekToBeginning
[Consumer clientId=consumer-group-2, groupId=group] Sending LIST_OFFSETS
request with header RequestHeader(apiKey=LIST_OFFSETS, apiVersion=10,
clientId=consumer-group-2, correlationId=31, headerVersion=2) and timeout 30000
to node 2: ListOffsetsRequestData(replicaId=-1, isolationLevel=0,
topics=[ListOffsetsTopic(name='topic',
partitions=[ListOffsetsPartition(partitionIndex=0, currentLeaderEpoch=0,
timestamp=-2)])], timeoutMs=30000) (org.apache.kafka.clients.NetworkClient:604)
[Consumer clientId=consumer-group-2, groupId=group] Resetting offset for
partition topic-0 to position FetchPosition{offset=0,
offsetEpoch=Optional.empty,
currentLeader=LeaderAndEpoch{leader=Optional[localhost:50970 (id: 2 rack: null
isFenced: false)], epoch=0}}.
(org.apache.kafka.clients.consumer.internals.SubscriptionState:451)
### The stale data before seekToBeginning is used by application thread, so the
next FetchRequest starts from offset 10.
[Consumer clientId=consumer-group-2, groupId=group] Added read_uncommitted
fetch request for partition topic-0 at position FetchPosition{offset=10,
offsetEpoch=Optional[0],
currentLeader=LeaderAndEpoch{leader=Optional[localhost:50970 (id: 2 rack: null
isFenced: false)], epoch=0}} to node localhost:50970 (id: 2 rack: null
isFenced: false) (org.apache.kafka.clients.consumer.internals.AbstractFetch:471)
[Consumer clientId=consumer-group-2, groupId=group] Sending FETCH request with
header RequestHeader(apiKey=FETCH, apiVersion=17, clientId=consumer-group-2,
correlationId=32, headerVersion=2) and timeout 30000 to node 2:
FetchRequestData(clusterId=null, replicaId=-1,
replicaState=ReplicaState(replicaId=-1, replicaEpoch=-1), maxWaitMs=500,
minBytes=1, maxBytes=52428800, isolationLevel=0, sessionId=784149360,
sessionEpoch=1, topics=[FetchTopic(topic='topic',
topicId=E2BqIjY8RU2mbcUClbcx3A, partitions=[FetchPartition(partition=0,
currentLeaderEpoch=0, fetchOffset=10, lastFetchedEpoch=-1, logStartOffset=-1,
partitionMaxBytes=1048576, replicaDirectoryId=AAAAAAAAAAAAAAAAAAAAAA)])],
forgottenTopicsData=[], rackId='') (org.apache.kafka.clients.NetworkClient:604)
{noformat}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
