jsancio commented on a change in pull request #10497:
URL: https://github.com/apache/kafka/pull/10497#discussion_r637228850
##########
File path:
core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala
##########
@@ -73,21 +76,33 @@ class BrokerMetadataListener(brokerId: Int,
/**
* Handle new metadata records.
*/
- override def handleCommits(lastOffset: Long, records:
util.List[ApiMessage]): Unit = {
- eventQueue.append(new HandleCommitsEvent(lastOffset, records))
+ override def handleCommit(reader: BatchReader[ApiMessageAndVersion]): Unit =
{
+ eventQueue.append(new HandleCommitsEvent(reader))
}
- // Visible for testing. It's useful to execute events synchronously
- private[metadata] def execCommits(lastOffset: Long, records:
util.List[ApiMessage]): Unit = {
- new HandleCommitsEvent(lastOffset, records).run()
+ // Visible for testing. It's useful to execute events synchronously in order
+ // to make tests deterministic
+ private[metadata] def execCommits(batch:
BatchReader.Batch[ApiMessageAndVersion]): Unit = {
+ new HandleCommitsEvent(BatchReader.singleton(batch)).run()
}
- class HandleCommitsEvent(lastOffset: Long,
- records: util.List[ApiMessage])
- extends EventQueue.FailureLoggingEvent(log) {
+ class HandleCommitsEvent(
+ reader: BatchReader[ApiMessageAndVersion]
+ ) extends EventQueue.FailureLoggingEvent(log) {
override def run(): Unit = {
+ try {
+ apply(reader.next())
Review comment:
Created https://issues.apache.org/jira/browse/KAFKA-12837 for this
suggestion.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
