KAFKA-15510: Fix follower's lastFetchedEpoch when fetch response has …

[chernyih]

…no record

A regression is introduced by #13843. When a fetch response has no record for a partition, validBytes is 0. In this case, we incorrectly set the last fetched epoch to logAppendInfo.lastLeaderEpoch.asScala which is Optional.empty. We should use currentFetchState.lastFetchedEpoch instead.

[rajinisivaram]

@chernyih Thanks for the PR, LGTM

[rajinisivaram]

@dajac Can you take a look as well, please? Thanks.

[divijvaidya]

Good catch!

Though, I am having a hard time wondering about the impact of this bug. The next fetch call from follower will provide (incorrectly) an empty epoch for last fetched epoch. In which case, if the follower was diverging, it will not correctly reconcile since leader will not know about the divergence. But if the follower was diverging, then is there a case where the new leader would have returned empty records?

I understand that we have added a unit test, but I would feel much more comfortable if we add a test which would demonstrates the impact of this bug on log reconciliation and hence, would catch similar bugs in future.

[rajinisivaram]

I couldn't think of a case where we might hit an actual truncation issue since divergence should have been detected in the first request after leader change, so will wait for @chernyih to respond. In any case, I think it makes sense to maintain the correct value for lastFetchedEpoch to avoid relying on timing assumptions that may not hold in future.

[divijvaidya]

In any case, I think it makes sense to maintain the correct value for lastFetchedEpoch to avoid relying on timing assumptions that may not hold in future.

Totally agree. The intention of my comment earlier was to gauge the potential impact of this bug and prevent re-occurrence in future. I completely agree that it should be fixed nevertheless.

[viktorsomogyi]

I also looked at how this lastFetchedEpoch is used in the next fetch but I don't think either it'd trigger truncation. Let's wait for @chernyih 's response. Since this is a small fix I think it'd be good to include it in the release.

[chernyih]

@rajinisivaram @divijvaidya @viktorsomogyi Thank you for the review. After reading the comments and thinking more about it, it won't impact truncation of fetch since if a leader has returned empty records, it means there is no divergence.

[chernyih]

The test failures do not look related. I compared with a few recent merged commit tests. For the failed tests that are not found in the recent commit tests, i ran those tests locally without failures.
https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka-pr/detail/PR-14457/1/tests/
https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka-pr/detail/PR-14457/2/tests

The second build was aborted. Didn't see any obvious error from the build log. I don't think the abort is related to this PR since the first build succeeded and saw another PR's build also was aborted. To be safe, we can rebuild. However, i couldn't find a way to rebuild.

From build log

Gradle Test Run :connect:runtime:test > Gradle Test Executor 83 > org.apache.kafka.connect.integration.BlockingConnectorTest > testBlockInSinkTaskStop SKIPPED
Daemon vm is shutting down... The daemon has exited normally or was terminated in response to a user interrupt.
----- End of the daemon log -----
FAILURE: Build failed with an exception.
* What went wrong:
Gradle build daemon disappeared unexpectedly (it may have been killed or may have crashed)

[rajinisivaram]

@chernyih Thanks for the update. Test failures not related, merging to trunk.