[GitHub] [kafka] mumrah commented on a diff in pull request #13344: MINOR: Replace BrokerMetadataListener with MetadataLoader

["mumrah (via GitHub)" <gi...@apache.org>]

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-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org