[GitHub] [kafka] cmccabe commented on a diff in pull request #12486: MINOR: BrokerMetadataSnapshotter must avoid exceeding batch size

[GitBox <gi...@apache.org>]

cmccabe commented on code in PR #12486:
URL: https://github.com/apache/kafka/pull/12486#discussion_r939446639


##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataSnapshotter.scala:
##########
@@ -24,18 +24,40 @@ import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
 import org.apache.kafka.server.common.ApiMessageAndVersion
 import org.apache.kafka.snapshot.SnapshotWriter
 
+import java.util.function.Consumer
+
 trait SnapshotWriterBuilder {
   def build(committedOffset: Long,
             committedEpoch: Int,
             lastContainedLogTime: Long): Option[SnapshotWriter[ApiMessageAndVersion]]
 }
 
+class RecordListConsumer(
+  val maxBatchSize: Int,
+  val writer: SnapshotWriter[ApiMessageAndVersion]
+) extends Consumer[java.util.List[ApiMessageAndVersion]] {
+  override def accept(messages: java.util.List[ApiMessageAndVersion]): Unit = {
+    var i = 0
+    while (i < messages.size()) {
+      writer.append(messages.subList(i, Math.min(i + maxBatchSize, messages.size())));
+      i += maxBatchSize
+    }
+  }
+}
+
 class BrokerMetadataSnapshotter(
   brokerId: Int,
   val time: Time,
   threadNamePrefix: Option[String],
   writerBuilder: SnapshotWriterBuilder
 ) extends Logging with MetadataSnapshotter {
+  /**
+   * The maximum number of records we will put in each batch. Do not change this unless
+   * you also change the equivalent constant in the Raft layer, and have considered the
+   * compatibility issues.
+   */
+  private val maxBatchSize = 1024

Review Comment:
   Good point. I added some more documentation here



##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataSnapshotter.scala:
##########
@@ -24,18 +24,40 @@ import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
 import org.apache.kafka.server.common.ApiMessageAndVersion
 import org.apache.kafka.snapshot.SnapshotWriter
 
+import java.util.function.Consumer
+
 trait SnapshotWriterBuilder {
   def build(committedOffset: Long,
             committedEpoch: Int,
             lastContainedLogTime: Long): Option[SnapshotWriter[ApiMessageAndVersion]]
 }
 
+class RecordListConsumer(

Review Comment:
   Added



-- 
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