junrao commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2052715308
##########
core/src/main/scala/kafka/raft/KafkaMetadataLog.scala:
##########
@@ -585,32 +585,62 @@ object KafkaMetadataLog extends Logging {
config: MetadataLogConfig,
nodeId: Int
): KafkaMetadataLog = {
- val props = new Properties()
- props.setProperty(TopicConfig.MAX_MESSAGE_BYTES_CONFIG,
config.maxBatchSizeInBytes.toString)
- props.setProperty(TopicConfig.SEGMENT_BYTES_CONFIG,
config.logSegmentBytes.toString)
- props.setProperty(TopicConfig.SEGMENT_MS_CONFIG,
config.logSegmentMillis.toString)
- props.setProperty(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG,
ServerLogConfigs.LOG_DELETE_DELAY_MS_DEFAULT.toString)
-
- // Disable time and byte retention when deleting segments
- props.setProperty(TopicConfig.RETENTION_MS_CONFIG, "-1")
- props.setProperty(TopicConfig.RETENTION_BYTES_CONFIG, "-1")
+ val props: Properties = settingLogProperties(config)
LogConfig.validate(props)
val defaultLogConfig = new LogConfig(props)
- if (config.logSegmentBytes < config.logSegmentMinBytes) {
- throw new InvalidConfigurationException(
- s"Cannot set ${MetadataLogConfig.METADATA_LOG_SEGMENT_BYTES_CONFIG}
below ${config.logSegmentMinBytes}: ${config.logSegmentBytes}"
- )
- } else if (defaultLogConfig.retentionMs >= 0) {
- throw new InvalidConfigurationException(
- s"Cannot set ${TopicConfig.RETENTION_MS_CONFIG} above -1:
${defaultLogConfig.retentionMs}."
- )
- } else if (defaultLogConfig.retentionSize >= 0) {
- throw new InvalidConfigurationException(
- s"Cannot set ${TopicConfig.RETENTION_BYTES_CONFIG} above -1:
${defaultLogConfig.retentionSize}."
- )
+ validateConfig(defaultLogConfig)
+
+ val metadataLog: KafkaMetadataLog = createMetadataLog(topicPartition,
topicId, dataDir, time, scheduler, config, nodeId, defaultLogConfig)
+
+ // Print a warning if users have overridden the internal config
+ if (config.logSegmentBytes() != KafkaRaftClient.MAX_BATCH_SIZE_BYTES) {
Review Comment:
Hmm, we don't need this if we add the constraint directly to
METADATA_LOG_SEGMENT_BYTES_CONFIG, right?
##########
raft/src/main/java/org/apache/kafka/raft/MetadataLogConfig.java:
##########
@@ -112,15 +108,15 @@ public class MetadataLogConfig {
* @param deleteDelayMillis The amount of time to wait before deleting a
file from the filesystem
*/
public MetadataLogConfig(int logSegmentBytes,
- int logSegmentMinBytes,
+ int internalLogSegmentMinBytes,
Review Comment:
Hmm, this should be renamed to internalLogSegmentBytes since it's no longer
the minimum, right?
##########
raft/src/main/java/org/apache/kafka/raft/MetadataLogConfig.java:
##########
@@ -85,14 +82,13 @@ public class MetadataLogConfig {
.define(METADATA_SNAPSHOT_MAX_INTERVAL_MS_CONFIG, LONG,
METADATA_SNAPSHOT_MAX_INTERVAL_MS_DEFAULT, atLeast(0), HIGH,
METADATA_SNAPSHOT_MAX_INTERVAL_MS_DOC)
.define(METADATA_LOG_DIR_CONFIG, STRING, null, null, HIGH,
METADATA_LOG_DIR_DOC)
.define(METADATA_LOG_SEGMENT_BYTES_CONFIG, INT,
METADATA_LOG_SEGMENT_BYTES_DEFAULT, atLeast(Records.LOG_OVERHEAD), HIGH,
METADATA_LOG_SEGMENT_BYTES_DOC)
- .defineInternal(METADATA_LOG_SEGMENT_MIN_BYTES_CONFIG, INT,
METADATA_LOG_SEGMENT_MIN_BYTES_DEFAULT, atLeast(Records.LOG_OVERHEAD), HIGH,
METADATA_LOG_SEGMENT_MIN_BYTES_DOC)
Review Comment:
We need to set the constraint for METADATA_LOG_SEGMENT_BYTES_CONFIG to be at
least 8MB.
Also, I thought the plan is to remove METADATA_LOG_SEGMENT_MIN_BYTES_CONFIG,
but add sth like METADATA_INTERNAL_LOG_SEGMENT_BYTES_CONFIG to match the design
in LogConfig?
##########
core/src/main/scala/kafka/raft/KafkaMetadataLog.scala:
##########
@@ -583,35 +584,72 @@ object KafkaMetadataLog extends Logging {
scheduler: Scheduler,
config: MetadataLogConfig
): KafkaMetadataLog = {
- val props = new Properties()
- props.setProperty(TopicConfig.MAX_MESSAGE_BYTES_CONFIG,
config.maxBatchSizeInBytes.toString)
- props.setProperty(TopicConfig.SEGMENT_BYTES_CONFIG,
config.logSegmentBytes.toString)
- props.setProperty(TopicConfig.SEGMENT_MS_CONFIG,
config.logSegmentMillis.toString)
- props.setProperty(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG,
ServerLogConfigs.LOG_DELETE_DELAY_MS_DEFAULT.toString)
-
- // Disable time and byte retention when deleting segments
- props.setProperty(TopicConfig.RETENTION_MS_CONFIG, "-1")
- props.setProperty(TopicConfig.RETENTION_BYTES_CONFIG, "-1")
+ val props: Properties = settingLogProperties(config)
LogConfig.validate(props)
val defaultLogConfig = new LogConfig(props)
- if (config.logSegmentBytes < config.logSegmentMinBytes) {
- throw new InvalidConfigurationException(
- s"Cannot set ${KRaftConfigs.METADATA_LOG_SEGMENT_BYTES_CONFIG} below
${config.logSegmentMinBytes}: ${config.logSegmentBytes}"
- )
- } else if (defaultLogConfig.retentionMs >= 0) {
- throw new InvalidConfigurationException(
- s"Cannot set ${TopicConfig.RETENTION_MS_CONFIG} above -1:
${defaultLogConfig.retentionMs}."
- )
- } else if (defaultLogConfig.retentionSize >= 0) {
- throw new InvalidConfigurationException(
- s"Cannot set ${TopicConfig.RETENTION_BYTES_CONFIG} above -1:
${defaultLogConfig.retentionSize}."
- )
+ validateConfig(config, defaultLogConfig)
+
+ val metadataLog: KafkaMetadataLog = createKafkaMetadataLog(topicPartition,
topicId, dataDir, time, scheduler, config, defaultLogConfig)
+
+ printWarningMessage(config, metadataLog)
+
+ // When recovering, truncate fully if the latest snapshot is after the log
end offset. This can happen to a follower
+ // when the follower crashes after downloading a snapshot from the leader
but before it could truncate the log fully.
+ metadataLog.truncateToLatestSnapshot()
+
+ metadataLog
+ }
+
+ private def printWarningMessage(config: MetadataLogConfig, metadataLog:
KafkaMetadataLog): Unit = {
+ // Print a warning if users have overridden the internal config
+ if (config.logSegmentMinBytes != KafkaRaftClient.MAX_BATCH_SIZE_BYTES) {
+ metadataLog.error(s"Overriding
${KRaftConfigs.METADATA_LOG_SEGMENT_MIN_BYTES_CONFIG} is only supported for
testing. Setting " +
+ s"this value too low may lead to an inability to write batches of
metadata records.")
}
+ }
+
+ // visible for testing
+ def internalApply(
Review Comment:
Hmm, why do we need `internalApply()`? MetadataLogConfig has both the
product and internal segment bytes configs and we could just pass both into
LogConfig in `apply()`, right?
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
