Unexpected called of `ElasticLogFileRecords.read` after partition reassignment

[Chillax-0v0]

stack:

ts=2023-11-29 12:09:21;thread_name=partition_op_2972;id=e14;is_daemon=true;priority=5;TCCL=jdk.internal.loader.ClassLoaders$AppClassLoader@251a69d7
    @kafka.log.streamaspect.ElasticLogFileRecords.read()
        at kafka.log.streamaspect.ElasticLogSegment.readAsync(ElasticLogSegment.scala:173)
        at kafka.log.streamaspect.ElasticLogSegment.read(ElasticLogSegment.scala:154)
        at kafka.log.LocalLog.$anonfun$read$2(LocalLog.scala:435)
        at kafka.log.LocalLog$.maybeHandleIOException(LocalLog.scala:820)
        at kafka.log.streamaspect.ElasticLog.maybeHandleIOException(ElasticLog.scala:112)
        at kafka.log.LocalLog.read(LocalLog.scala:404)
        at kafka.log.LocalLog.convertToOffsetMetadataOrThrow(LocalLog.scala:384)
        at kafka.log.UnifiedLog.convertToOffsetMetadataOrThrow(UnifiedLog.scala:1385)
        at kafka.log.UnifiedLog.fetchHighWatermarkMetadata(UnifiedLog.scala:468)
        at kafka.log.UnifiedLog.maybeIncrementHighWatermark(UnifiedLog.scala:423)
        at kafka.cluster.Partition.maybeIncrementLeaderHW(Partition.scala:1067)
        at kafka.cluster.Partition.$anonfun$makeLeader$1(Partition.scala:731)
        at kafka.cluster.Partition.makeLeader(Partition.scala:633)
        at kafka.server.ReplicaManager.$anonfun$applyLocalLeadersDelta$3(ReplicaManager.scala:2510)
        at scala.Option.foreach(Option.scala:437)
        at kafka.server.ReplicaManager.$anonfun$applyLocalLeadersDelta$2(ReplicaManager.scala:2507)
        at kafka.server.ReplicaManager.$anonfun$applyLocalLeadersDelta$2$adapted(ReplicaManager.scala:2506)
        at kafka.utils.Implicits$MapExtensionMethods$.$anonfun$forKeyValue$1(Implicits.scala:62)
        at scala.collection.mutable.HashMap$Node.foreachEntry(HashMap.scala:633)
        at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:499)
        at kafka.server.ReplicaManager.applyLocalLeadersDelta(ReplicaManager.scala:2506)
        at kafka.server.ReplicaManager.$anonfun$asyncApplyDelta$12(ReplicaManager.scala:2454)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
        at java.lang.Thread.run(Thread.java:833)

[mooc9988]

Direct cause: After creating log, HW should be recovered from HW offsetCheckpoints. It can be referred in kafka.cluster.Partition#createLog.

Then it will use kafka.log.UnifiedLog#updateHighWatermark:

This issue may be due to missing HW checkpoint or wrong HW checkpoint.

[mooc9988]

Root cause: ElasticLogManager was enabled after the construction of logManager and ReplicaManager. As a result, all checkpoints were read from local files instead of the data in streams.

The construction of logManager or ReplicaManager will refer to checkpoints, like recoveryPointCheckpoints. Since ElasticLogManager was not enabled, it will always use local checkpoint files.

[mooc9988]

In kafka.server.ReplicaManager#applyLocalLeadersDelta, leading logs are generated and added to ElasticLogManager individually. However, it only extracts 'offsetCheckpoints' after the creation of the first leading log. Consequently, other leading logs encounter issues in obtaining checkpoints from 'offsetCheckpoints'.