mumrah commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2109579630
##########
core/src/main/scala/kafka/raft/KafkaMetadataLog.scala:
##########
@@ -639,12 +637,6 @@ object KafkaMetadataLog extends Logging {
nodeId
)
- // Print a warning if users have overridden the internal config
- if (config.logSegmentMinBytes != KafkaRaftClient.MAX_BATCH_SIZE_BYTES) {
Review Comment:
Should we keep the error message if INTERNAL_SEGMENT_BYTES_CONFIG has been
set? This was added to help operators avoid setting these testing-only configs
in a production environment.
##########
core/src/main/scala/kafka/raft/KafkaMetadataLog.scala:
##########
@@ -599,11 +601,7 @@ object KafkaMetadataLog extends Logging {
LogConfig.validate(props)
val defaultLogConfig = new LogConfig(props)
- if (config.logSegmentBytes < config.logSegmentMinBytes) {
Review Comment:
IIUC, this logic has been replaced by the config validation system.
SEGMENT_BYTES_CONFIG has minimum of 1 MB but can be overridden by
INTERNAL_SEGMENT_BYTES_CONFIG which has no minimum.
##########
raft/src/main/java/org/apache/kafka/raft/MetadataLogConfig.java:
##########
@@ -80,72 +78,55 @@ public class MetadataLogConfig {
"controller should write no-op records to the metadata partition.
If the value is 0, no-op records " +
"are not appended to the metadata partition. The default value is
" + METADATA_MAX_IDLE_INTERVAL_MS_DEFAULT;
+ public static final String INTERNAL_MAX_BATCH_SIZE_IN_BYTES_CONFIG =
"internal.max.batch.size.in.bytes";
+ public static final String INTERNAL_MAX_BATCH_SIZE_IN_BYTES_DOC = "The
largest record batch size allowed in the metadata log, only for testing.";
+
+ public static final String INTERNAL_MAX_FETCH_SIZE_IN_BYTES_CONFIG =
"internal.max.fetch.size.in.bytes";
+ public static final String INTERNAL_MAX_FETCH_SIZE_IN_BYTES_DOC = "The
maximum number of bytes to read when fetching from the metadata log, only for
testing.";
+
+ public static final String INTERNAL_DELETE_DELAY_MILLIS_CONFIG =
"internal.delete.delay.millis";
+ public static final String INTERNAL_DELETE_DELAY_MILLIS_DOC = "The amount
of time to wait before deleting a file from the filesystem, only for testing.";
+
Review Comment:
Shouldn't these configs have "metadata" in there somewhere in the name? From
what I can tell, these are only used by KafkaMetadataLog.
##########
raft/src/main/java/org/apache/kafka/raft/MetadataLogConfig.java:
##########
@@ -80,72 +78,55 @@ public class MetadataLogConfig {
"controller should write no-op records to the metadata partition.
If the value is 0, no-op records " +
"are not appended to the metadata partition. The default value is
" + METADATA_MAX_IDLE_INTERVAL_MS_DEFAULT;
+ public static final String INTERNAL_MAX_BATCH_SIZE_IN_BYTES_CONFIG =
"internal.max.batch.size.in.bytes";
+ public static final String INTERNAL_MAX_BATCH_SIZE_IN_BYTES_DOC = "The
largest record batch size allowed in the metadata log, only for testing.";
+
+ public static final String INTERNAL_MAX_FETCH_SIZE_IN_BYTES_CONFIG =
"internal.max.fetch.size.in.bytes";
+ public static final String INTERNAL_MAX_FETCH_SIZE_IN_BYTES_DOC = "The
maximum number of bytes to read when fetching from the metadata log, only for
testing.";
+
+ public static final String INTERNAL_DELETE_DELAY_MILLIS_CONFIG =
"internal.delete.delay.millis";
+ public static final String INTERNAL_DELETE_DELAY_MILLIS_DOC = "The amount
of time to wait before deleting a file from the filesystem, only for testing.";
+
public static final ConfigDef CONFIG_DEF = new ConfigDef()
.define(METADATA_SNAPSHOT_MAX_NEW_RECORD_BYTES_CONFIG, LONG,
METADATA_SNAPSHOT_MAX_NEW_RECORD_BYTES, atLeast(1), HIGH,
METADATA_SNAPSHOT_MAX_NEW_RECORD_BYTES_DOC)
.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)
+ .define(METADATA_LOG_SEGMENT_BYTES_CONFIG, INT,
METADATA_LOG_SEGMENT_BYTES_DEFAULT, atLeast(8 * 1024 * 1024), HIGH,
METADATA_LOG_SEGMENT_BYTES_DOC)
.define(METADATA_LOG_SEGMENT_MILLIS_CONFIG, LONG,
METADATA_LOG_SEGMENT_MILLIS_DEFAULT, null, HIGH,
METADATA_LOG_SEGMENT_MILLIS_DOC)
.define(METADATA_MAX_RETENTION_BYTES_CONFIG, LONG,
METADATA_MAX_RETENTION_BYTES_DEFAULT, null, HIGH,
METADATA_MAX_RETENTION_BYTES_DOC)
.define(METADATA_MAX_RETENTION_MILLIS_CONFIG, LONG,
METADATA_MAX_RETENTION_MILLIS_DEFAULT, null, HIGH,
METADATA_MAX_RETENTION_MILLIS_DOC)
- .define(METADATA_MAX_IDLE_INTERVAL_MS_CONFIG, INT,
METADATA_MAX_IDLE_INTERVAL_MS_DEFAULT, atLeast(0), LOW,
METADATA_MAX_IDLE_INTERVAL_MS_DOC);
+ .define(METADATA_MAX_IDLE_INTERVAL_MS_CONFIG, INT,
METADATA_MAX_IDLE_INTERVAL_MS_DEFAULT, atLeast(0), LOW,
METADATA_MAX_IDLE_INTERVAL_MS_DOC)
+ .defineInternal(INTERNAL_METADATA_LOG_SEGMENT_BYTES_CONFIG, INT,
null, null, LOW, INTERNAL_METADATA_LOG_SEGMENT_BYTES_DOC)
+ .defineInternal(INTERNAL_MAX_BATCH_SIZE_IN_BYTES_CONFIG, INT,
KafkaRaftClient.MAX_BATCH_SIZE_BYTES, null, LOW,
INTERNAL_MAX_BATCH_SIZE_IN_BYTES_DOC)
+ .defineInternal(INTERNAL_MAX_FETCH_SIZE_IN_BYTES_CONFIG, INT,
KafkaRaftClient.MAX_FETCH_SIZE_BYTES, null, LOW,
INTERNAL_MAX_FETCH_SIZE_IN_BYTES_DOC)
+ .defineInternal(INTERNAL_DELETE_DELAY_MILLIS_CONFIG, LONG,
ServerLogConfigs.LOG_DELETE_DELAY_MS_DEFAULT, null, LOW,
INTERNAL_DELETE_DELAY_MILLIS_DOC);
private final int logSegmentBytes;
- private final int logSegmentMinBytes;
+ private final Integer internalSegmentBytes;
private final long logSegmentMillis;
private final long retentionMaxBytes;
private final long retentionMillis;
- private final int maxBatchSizeInBytes;
- private final int maxFetchSizeInBytes;
- private final long deleteDelayMillis;
-
- /**
- * Configuration for the metadata log
- * @param logSegmentBytes The maximum size of a single metadata log file
- * @param logSegmentMinBytes The minimum size of a single metadata log file
- * @param logSegmentMillis The maximum time before a new metadata log file
is rolled out
- * @param retentionMaxBytes The size of the metadata log and snapshots
before deleting old snapshots and log files
- * @param retentionMillis The time to keep a metadata log file or snapshot
before deleting it
- * @param maxBatchSizeInBytes The largest record batch size allowed in the
metadata log
- * @param maxFetchSizeInBytes The maximum number of bytes to read when
fetching from the metadata log
- * @param deleteDelayMillis The amount of time to wait before deleting a
file from the filesystem
- */
- public MetadataLogConfig(int logSegmentBytes,
- int logSegmentMinBytes,
- long logSegmentMillis,
- long retentionMaxBytes,
- long retentionMillis,
- int maxBatchSizeInBytes,
- int maxFetchSizeInBytes,
- long deleteDelayMillis) {
- this.logSegmentBytes = logSegmentBytes;
- this.logSegmentMinBytes = logSegmentMinBytes;
- this.logSegmentMillis = logSegmentMillis;
- this.retentionMaxBytes = retentionMaxBytes;
- this.retentionMillis = retentionMillis;
- this.maxBatchSizeInBytes = maxBatchSizeInBytes;
- this.maxFetchSizeInBytes = maxFetchSizeInBytes;
- this.deleteDelayMillis = deleteDelayMillis;
- }
+ private final int internalMaxBatchSizeInBytes;
+ private final int internalMaxFetchSizeInBytes;
+ private final long internalDeleteDelayMillis;
public MetadataLogConfig(AbstractConfig config) {
this.logSegmentBytes =
config.getInt(METADATA_LOG_SEGMENT_BYTES_CONFIG);
- this.logSegmentMinBytes =
config.getInt(METADATA_LOG_SEGMENT_MIN_BYTES_CONFIG);
+ this.internalSegmentBytes =
config.getInt(INTERNAL_METADATA_LOG_SEGMENT_BYTES_CONFIG);
this.logSegmentMillis =
config.getLong(METADATA_LOG_SEGMENT_MILLIS_CONFIG);
this.retentionMaxBytes =
config.getLong(METADATA_MAX_RETENTION_BYTES_CONFIG);
this.retentionMillis =
config.getLong(METADATA_MAX_RETENTION_MILLIS_CONFIG);
- this.maxBatchSizeInBytes = KafkaRaftClient.MAX_BATCH_SIZE_BYTES;
- this.maxFetchSizeInBytes = KafkaRaftClient.MAX_FETCH_SIZE_BYTES;
- this.deleteDelayMillis = ServerLogConfigs.LOG_DELETE_DELAY_MS_DEFAULT;
+ this.internalMaxBatchSizeInBytes =
config.getInt(INTERNAL_MAX_BATCH_SIZE_IN_BYTES_CONFIG);
+ this.internalMaxFetchSizeInBytes =
config.getInt(INTERNAL_MAX_FETCH_SIZE_IN_BYTES_CONFIG);
+ this.internalDeleteDelayMillis =
config.getLong(INTERNAL_DELETE_DELAY_MILLIS_CONFIG);
}
public int logSegmentBytes() {
return logSegmentBytes;
}
-
- public int logSegmentMinBytes() {
- return logSegmentMinBytes;
+
+ public Integer internalSegmentBytes() {
Review Comment:
`OptionalInt` might be nicer here.
--
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
