KAFKA-9087 Replace log high watermark by future log high watermark wh…

[chia7712]

https://issues.apache.org/jira/browse/KAFKA-9087

[2019-06-11 17:58:46,304] ERROR [ReplicaAlterLogDirsThread-1]: Error due to (kafka.server.ReplicaAlterLogDirsThread)
 org.apache.kafka.common.KafkaException: Error processing data for partition metrics_timers-35 offset 4224887
 at kafka.server.AbstractFetcherThread.$anonfun$processFetchRequest$7(AbstractFetcherThread.scala:342)
 at scala.Option.foreach(Option.scala:274)
 at kafka.server.AbstractFetcherThread.$anonfun$processFetchRequest$6(AbstractFetcherThread.scala:300)
 at kafka.server.AbstractFetcherThread.$anonfun$processFetchRequest$6$adapted(AbstractFetcherThread.scala:299)
 at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
 at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
 at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
 at kafka.server.AbstractFetcherThread.$anonfun$processFetchRequest$5(AbstractFetcherThread.scala:299)
 at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
 at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:251)
 at kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:299)
 at kafka.server.AbstractFetcherThread.$anonfun$maybeFetch$3(AbstractFetcherThread.scala:132)
 at kafka.server.AbstractFetcherThread.$anonfun$maybeFetch$3$adapted(AbstractFetcherThread.scala:131)
 at scala.Option.foreach(Option.scala:274)
 at kafka.server.AbstractFetcherThread.maybeFetch(AbstractFetcherThread.scala:131)
 at kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:113)
 at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:82)
 Caused by: java.lang.IllegalStateException: Offset mismatch for the future replica metrics_timers-35: fetched offset = 4224887, log end offset = 0.
 at kafka.server.ReplicaAlterLogDirsThread.processPartitionData(ReplicaAlterLogDirsThread.scala:107)
 at kafka.server.AbstractFetcherThread.$anonfun$processFetchRequest$7(AbstractFetcherThread.scala:311)

The race condition of processing LeaderAndIsrRequest and AlterReplicaLogDirsRequest causes above error (on V2 message format). Also, the error can be reproduced easily on V1 since there is no epoch cache. The ReplicaAlterLogDirsThread checks the offset of “future log” rather than “log. Hence, here is my two cents, we can replace log.highWatermark by futureLog.highWatermark to resolve this issue. I tested it on our cluster and it works well (on both V1 and V2).

Committer Checklist (excluded from commit message)

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

[ijuma]

Is it possible to add a test?

[jolshan]

So the error occurs when the fetch offset is not the futureLog's log end offset.

Just curious -- is this because the log's highwatermark is lower or higher than the futureLog's?

It seems that the value we put into the InitialFetchState is the first offset we fetch from, so just wanted to clarify the race and why this fixes the issue. (Not doubting you, just trying to understand 😄 )

[junrao]

@jolshan : It seems that this is a regression introduced in #6841. When a future log is just created, the current log's HWM could be higher than the future replica's HWM (since it's bounded by its log end offset). When the future replica starts fetching, we want it to fetch from its local HWM, not the current log's HWM. Otherwise, there will be holes in the future log.

[chia7712]

is this because the log's highwatermark is lower or higher than the futureLog's?

higher, and thanks to @junrao for the great explanation.

[junrao]

@chia7712 : Thanks for the PR. Just one minor comment.

[chia7712]

@junrao Please take a look at line#258

[junrao]

@chia7712 : Thanks for the updated PR. A couple of more comments.

[junrao]

@chia7712 : Thanks for the updated PR. LGTM if the tests pass.

[chia7712]

unrelated error

Build / JDK 11 and Scala 2.13 / kafka.admin.ReassignPartitionsIntegrationTest.testAlterReassignmentThrottle(String).quorum=kraft | 26 秒 | 1
Build / JDK 11 and Scala 2.13 / kafka.admin.TopicCommandIntegrationTest.testTopicWithCollidingCharDeletionAndCreateAgain(String).quorum=kraft
[Build / JDK 11 and Scala 2.13 / org.apache.kafka.connect.integration.ConnectorTopicsIntegrationTest.testGetActiveTopics](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-13075/7/testReport/org.apache.kafka.connect.integration/ConnectorTopicsIntegrationTest/Build___JDK_11_and_Scala_2_13___testGetActiveTopics/)

will merge it