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

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

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


##########
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:
   Maybe add a quick javadoc of what this class does? Also, maybe a more descriptive name (BatchingRecordListConsumer?)



##########
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:
   Can we reference a constant for this special value?



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