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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
