KAFKA-15080; Fetcher's lag never set when partition is idle

[dajac]

The PartitionFetchState's lag field is set to None when the state is created and it is updated when bytes are received for a partition. For idle partitions (newly created or not), the lag is never updated because validBytes > 0 is never true. As a side effect, the partition is considered out-of-sync and could be incorrectly throttled.

This patch really makes me wonder if keeping the lag in the PartitionFetchState is really necessary. At the end of the day, we only use is to compute isReplicaInSync because the metrics rely on the the FetcherLagMetrics. We could just maintain a boolean for this purpose. I chose the easy path in this patch.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[divijvaidya]

The changes looks good to me but I am afraid that the absence of lag field in InitialFetchState is causing problems at other places as well such as

kafka/core/src/main/scala/kafka/server/AbstractFetcherManager.scala

Line 74 in c0cb8dd

val initialFetchState = InitialFetchState(currentFetchState.topicId, thread.leader.brokerEndPoint(),

where we ideally should be copying the lag of existing partition states to the new one when migrating partitions from one thread to another.

This makes me wonder if we should either include lag in InitialState or follow the suggestion you provided to not rely on lag.

[dajac]

This is exactly why I check currentFetchState.lag.isEmpty to ensure that it gets initialised in all cases on the first fetch. I would not include lag in InitialState because this is also what we use when a partition is added by the replica manager and the lag does not have any sense in this context. In the case you mentioned, the lag will be set on the first fetch.

[dajac]

I am playing around with my suggestion but this is way more invasive than the current proposed change so I wonder if we should start by merging a simple fix while I work on a long term solution...

[divijvaidya]

LGTM! Let's merge this PR and continue discussion on the other fix.

In the case you mentioned, the lag will be set on the first fetch.
Yes, but we already have that information. We should be reusing it from older state. Why wait for next fetch?

[dajac]

Yeah, I agree but InitialFetchState is not the right way here. Instead, we should refactor resizeThreadPool to not use it but I am not sure that it is worth given than resizing is a rare event.

[hachikuji]

LGTM