KAFKA-9724 Newer clients not always sending fetch request to older brokers

[mumrah]

We had a similar case previously with KAFKA-8422 (#6806) where we would skip the validation step if the broker was on a version older than 2.3.

This PR makes a similar change on the prepareFetchRequest side. If the broker is older than 2.3, we will skip the transition to AWAITING_VALIDATION but also we will clear that state if it had been set by a call to seekUnvalidated.

This PR adds a check for an appropriate version of OFFSETS_FOR_LEADER_EPOCH before entering the validation state in the consumer. If the broker does not support version 3+ of this API, we skip validation.

Also cleaned up the logic behind updating last-seen leader epochs. Now, if the leader epoch is null, we will not update it unless we are processing a metadata response. This prevents the client from learning about an epoch before it knows anything about the leader.

[mumrah]

@hachikuji this change might be too superficial. It should fix the problem when talking to older brokers, but ideally we shouldn't enter the AWAITING_VALIDATION state at all if the broker is older than 2.3. Doing that would probably require pushing down a flag into the TopicPartitionState which would be a bigger change. WTDY?

[ijuma]

ok to test

[hachikuji]

Note that FetcherTest.testFetcherConcurrency is hanging.

[mumrah]

I removed this since at this point the two callers have different needs for updating the epoch and adding a flag felt pretty kludgy

[mumrah]

I wonder, do we really need this call here? If the leader is not present the epoch shouldn't be present either -- right? If that's the case, then the call to maybeValidatePosition will short circuit

        private boolean maybeValidatePosition(Metadata.LeaderAndEpoch currentLeaderAndEpoch) {
            if (this.fetchState.equals(FetchStates.AWAIT_RESET)) {
                return false;
            }

            if (!currentLeaderAndEpoch.leader.isPresent() && !currentLeaderAndEpoch.epoch.isPresent()) {
                return false;
            }

            if (position != null && !position.currentLeader.equals(currentLeaderAndEpoch)) {
                FetchPosition newPosition = new FetchPosition(position.offset, position.offsetEpoch, currentLeaderAndEpoch);
                validatePosition(newPosition);
                preferredReadReplica = null;
            }
            return this.fetchState.equals(FetchStates.AWAIT_VALIDATION);
        }

[mumrah]

Oh, actually looking at the javadoc for LeaderAndEpoch, I see

It is also possible that we know of the leader epoch, but not the leader when it is derived from an external source (e.g. a committed offset).

Also in Metadata, we do return a LeaderAndEpoch with the last-seen epoch, but no leader if the metadata is stale. So, I guess it makes sense to keep this call in maybeValidatePositionForCurrentLeader

[hachikuji]

LGTM (assuming tests pass after checkstyle is fixed)

[ijuma]

Do we think this is safe to backport or is it best to stick with 2.6.0?

[hachikuji]

@ijuma I think it is safe. I was planning to backport to 2.5.

[ijuma]

Sounds good.

[hachikuji]

retest this please

[mumrah]

retest this please

[mumrah]

Test failures are known flaky tests

• EosBetaUpgradeIntegrationTest
• PreferredReplicaLeaderElectionCommandTest#testBasicPreferredReplicaElection