hachikuji commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r907665218
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala:
##########
@@ -117,26 +117,30 @@ class BrokerMetadataListener(
} finally {
reader.close()
}
- _publisher.foreach(publish)
- // If we detected a change in metadata.version, generate a local snapshot
- val metadataVersionChanged = Option(_delta.featuresDelta()).exists {
featuresDelta =>
- featuresDelta.metadataVersionChange().isPresent
+ _bytesSinceLastSnapshot = _bytesSinceLastSnapshot + results.numBytes
+ if (_publisher.nonEmpty && shouldSnapshot()) {
Review Comment:
Perhaps we could move the publisher check into `metadataVersionChanged()`?
My understanding is that `metadataVersionChanged` will always returns true
until we have applied the first delta, so that would make the dependence a
little clearer. Do I have that right?
A consequence of changing that is that we could publish snapshots before the
publisher has started if enough bytes have accumulated, which was the prior
behavior. That would mean that we wouldn't need to update
`StartPublishingEvent` below. Do you see any issues with that?
--
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
