mumrah commented on code in PR #13344:
URL: https://github.com/apache/kafka/pull/13344#discussion_r1132741472
##########
core/src/main/scala/kafka/server/BrokerServer.scala:
##########
@@ -362,16 +326,13 @@ class BrokerServer(
config.brokerSessionTimeoutMs / 2 // KAFKA-14392
)
lifecycleManager.start(
- () => metadataListener.highestMetadataOffset,
+ () => sharedServer.loader.lastAppliedOffset(),
Review Comment:
Just want to check that the semantics are the same here (since the
terminology and code is slightly different). Before, we were sending
`highestMetadataOffset` which is updated as we replay messages:
```scala
_highestOffset = lastCommittedOffset.getOrElse(batch.baseOffset() +
index)
try {
delta.replay(messageAndVersion.message())
} catch {
```
We don't actually call the publishers until after we have replayed the whole
batch.
With the new metadata loader infrastructure, we update the
`lastAppliedOffset` after we have applied the whole batch and called the
publishers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
