You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/08/05 06:55:16 UTC
[GitHub] [kafka] cmccabe opened a new pull request, #12486: MINOR: BrokerMetadataSnapshotter must avoid exceeding batch size
cmccabe opened a new pull request, #12486:
URL: https://github.com/apache/kafka/pull/12486
BrokerMetadataSnapshotter should split up record lists that exceed the batch size.
--
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
[GitHub] [kafka] cmccabe commented on a diff in pull request #12486: MINOR: BrokerMetadataSnapshotter must avoid exceeding batch size
Posted by 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
[GitHub] [kafka] mumrah commented on a diff in pull request #12486: MINOR: BrokerMetadataSnapshotter must avoid exceeding batch size
Posted by 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
[GitHub] [kafka] cmccabe merged pull request #12486: MINOR: BrokerMetadataSnapshotter must avoid exceeding batch size
Posted by GitBox <gi...@apache.org>.
cmccabe merged PR #12486:
URL: https://github.com/apache/kafka/pull/12486
--
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
[GitHub] [kafka] cmccabe commented on pull request #12486: MINOR: BrokerMetadataSnapshotter must avoid exceeding batch size
Posted by GitBox <gi...@apache.org>.
cmccabe commented on PR #12486:
URL: https://github.com/apache/kafka/pull/12486#issuecomment-1206848794
> Thanks @cmccabe, just one main question -- is it okay to break up an atomic batch into two Lists when writing to a snapshot? Do we ever rely on reading an atomic batch of messages as one List when reading snapshots? E.g., what happens if we're dealing with a topic with 2000 partitions.
No, we do not rely on this in the snapshot. The decision of what batches to use in the snapshot doesn't change semantics.
In the log (as opposed to snapshot) we care about batching because snapshots are atomically committed, so if the controller fails midway into writing a multibatch operation we have to think about what happens next (probably a partial commit of whatever was going on)
--
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