Skip to content

MINOR: Raft max batch size needs to propagate to log config #10256

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 3 commits into from
Mar 4, 2021
Merged

MINOR: Raft max batch size needs to propagate to log config #10256

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c10f840
MINOR: Raft max batch size needs to propagate to log config
hachikuji Mar 4, 2021
dfdd4a3
Test cases for max batch size
hachikuji Mar 4, 2021
fad6688
Remove second `apply` and add helper for snapshot recovery
hachikuji Mar 4, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 44 additions & 20 deletions core/src/main/scala/kafka/raft/KafkaMetadataLog.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,29 +16,20 @@
*/
package kafka.raft

import java.nio.file.Files
import java.nio.file.NoSuchFileException
import java.util.NoSuchElementException
import java.util.Optional
import java.io.File
import java.nio.file.{Files, NoSuchFileException}
import java.util.concurrent.ConcurrentSkipListSet
import java.util.{NoSuchElementException, Optional, Properties}

import kafka.log.{AppendOrigin, Log, SnapshotGenerated, LogOffsetSnapshot}
import kafka.server.{FetchHighWatermark, FetchLogEnd}
import kafka.api.ApiVersion
import kafka.log.{AppendOrigin, Log, LogConfig, LogOffsetSnapshot, SnapshotGenerated}
import kafka.server.{BrokerTopicStats, FetchHighWatermark, FetchLogEnd, LogDirFailureChannel}
import kafka.utils.Scheduler
import org.apache.kafka.common.record.{MemoryRecords, Records}
import org.apache.kafka.common.utils.Time
import org.apache.kafka.common.{KafkaException, TopicPartition}
import org.apache.kafka.raft.Isolation
import org.apache.kafka.raft.LogAppendInfo
import org.apache.kafka.raft.LogFetchInfo
import org.apache.kafka.raft.LogOffsetMetadata
import org.apache.kafka.raft.OffsetAndEpoch
import org.apache.kafka.raft.OffsetMetadata
import org.apache.kafka.raft.ReplicatedLog
import org.apache.kafka.snapshot.FileRawSnapshotReader
import org.apache.kafka.snapshot.FileRawSnapshotWriter
import org.apache.kafka.snapshot.RawSnapshotReader
import org.apache.kafka.snapshot.RawSnapshotWriter
import org.apache.kafka.snapshot.SnapshotPath
import org.apache.kafka.snapshot.Snapshots
import org.apache.kafka.raft.{Isolation, LogAppendInfo, LogFetchInfo, LogOffsetMetadata, OffsetAndEpoch, OffsetMetadata, ReplicatedLog}
import org.apache.kafka.snapshot.{FileRawSnapshotReader, FileRawSnapshotWriter, RawSnapshotReader, RawSnapshotWriter, SnapshotPath, Snapshots}

import scala.compat.java8.OptionConverters._

Expand Down Expand Up @@ -297,10 +288,43 @@ final class KafkaMetadataLog private (
}

object KafkaMetadataLog {

def apply(
topicPartition: TopicPartition,
dataDir: File,
time: Time,
scheduler: Scheduler,
maxBatchSizeInBytes: Int,
maxFetchSizeInBytes: Int
): KafkaMetadataLog = {
val props = new Properties()
props.put(LogConfig.MaxMessageBytesProp, maxBatchSizeInBytes.toString)
props.put(LogConfig.MessageFormatVersionProp, ApiVersion.latestVersion.toString)

LogConfig.validateValues(props)
val defaultLogConfig = LogConfig(props)

val log = Log(

@mumrah mumrah Mar 4, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍 Makes sense to move this here since KafkaMetadataLog fully owns the Log's lifecycle

dir = dataDir,
config = defaultLogConfig,
logStartOffset = 0L,
recoveryPoint = 0L,
scheduler = scheduler,
brokerTopicStats = new BrokerTopicStats,
time = time,
maxProducerIdExpirationMs = Int.MaxValue,
producerIdExpirationCheckIntervalMs = Int.MaxValue,
logDirFailureChannel = new LogDirFailureChannel(5),
keepPartitionMetadataFile = false
)

KafkaMetadataLog(log, topicPartition, maxFetchSizeInBytes)
}

private def apply(
Comment thread
hachikuji marked this conversation as resolved.
Outdated
log: Log,
topicPartition: TopicPartition,
maxFetchSizeInBytes: Int = 1024 * 1024
maxFetchSizeInBytes: Int
): KafkaMetadataLog = {
val snapshotIds = new ConcurrentSkipListSet[OffsetAndEpoch]()
// Scan the log directory; deleting partial snapshots and remembering immutable snapshots
Expand Down
29 changes: 9 additions & 20 deletions core/src/main/scala/kafka/raft/RaftManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,9 +22,9 @@ import java.util
import java.util.OptionalInt
import java.util.concurrent.CompletableFuture

import kafka.log.{Log, LogConfig, LogManager}
import kafka.log.Log
import kafka.raft.KafkaRaftManager.RaftIoThread
import kafka.server.{BrokerTopicStats, KafkaConfig, LogDirFailureChannel, MetaProperties}
import kafka.server.{KafkaConfig, MetaProperties}
import kafka.utils.timer.SystemTimer
import kafka.utils.{KafkaScheduler, Logging, ShutdownableThread}
import org.apache.kafka.clients.{ApiVersions, ClientDnsLookup, ManualMetadataUpdater, NetworkClient}
Expand Down Expand Up @@ -220,25 +220,14 @@ class KafkaRaftManager[T](
}

private def buildMetadataLog(): KafkaMetadataLog = {
val defaultProps = LogConfig.extractLogConfigMap(config)

@mumrah mumrah Mar 4, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see we are no longer calling this, but rather explicitly populating the log properties. Are there other log configs we might want to expose?

@hachikuji hachikuji Mar 4, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seemed best for now to not allow overrides since this is a new usage of Log and we haven't had time to understand the impact of all of the configurations for this usage. Most of them are probably safe, but others do not even make sense (e.g. the retention settings). Let me open a JIRA so that we can consider which configs we want to expose for the metadata log and how we want to expose them.

@hachikuji hachikuji Mar 4, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I filed this JIRA: https://issues.apache.org/jira/browse/KAFKA-12423.

LogConfig.validateValues(defaultProps)
val defaultLogConfig = LogConfig(defaultProps)

val log = Log(
dir = dataDir,
config = defaultLogConfig,
logStartOffset = 0L,
recoveryPoint = 0L,
scheduler = scheduler,
brokerTopicStats = new BrokerTopicStats,
time = time,
maxProducerIdExpirationMs = config.transactionalIdExpirationMs,
producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs,
logDirFailureChannel = new LogDirFailureChannel(5),
keepPartitionMetadataFile = config.usesTopicId
KafkaMetadataLog(
topicPartition,
dataDir,
time,
scheduler,
maxBatchSizeInBytes = KafkaRaftClient.MAX_BATCH_SIZE_BYTES,
maxFetchSizeInBytes = KafkaRaftClient.MAX_FETCH_SIZE_BYTES
)

KafkaMetadataLog(log, topicPartition)
}

private def buildNetworkClient(): NetworkClient = {
Expand Down
Loading