You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "m1a2st (via GitHub)" <gi...@apache.org> on 2025/04/04 09:29:02 UTC
[PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
m1a2st opened a new pull request, #19371:
URL: https://github.com/apache/kafka/pull/19371
The main reason is that we forgot setting the `TopicConfig.SEGMENT_BYTES_CONFIG` at least to `1024 * 1024`, thus addressed it, and add a test for it.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2086165049
##########
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:
@m1a2st What do you think about the `METADATA_INTERNAL_LOG_SEGMENT_BYTES_CONFIG`? This PR currently add only `INTERNAL_LOG_SEGMENT_BYTES_CONFIG` to represent server-level log config. That means users can't configure "small" metadata log without impacting other logs.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2135853739
##########
core/src/main/scala/kafka/raft/KafkaMetadataLog.scala:
##########
@@ -588,9 +588,11 @@ object KafkaMetadataLog extends Logging {
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)
Review Comment:
this is a critical bug - open https://issues.apache.org/jira/browse/KAFKA-19392 to fix it
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #19371:
URL: https://github.com/apache/kafka/pull/19371#issuecomment-2905909017
No, I think the fail tes are not related, but the main reason is the hatset order is different, I will file a PR to fix it.
```
org.opentest4j.AssertionFailedError: expected: <[CoordinatorRecord(key=ShareGroupStatePartitionMetadataKey(groupId='groupId'), value=ApiMessageAndVersion(ShareGroupStatePartitionMetadataValue(initializingTopics=[], initializedTopics=[TopicPartitionsInfo(topicId=wO0yz3DfTaCC9fnfn8maXw, topicName='t1', partitions=[1, 0])], deletingTopics=[]) at version 0))]> but was: <[CoordinatorRecord(key=ShareGroupStatePartitionMetadataKey(groupId='groupId'), value=ApiMessageAndVersion(ShareGroupStatePartitionMetadataValue(initializingTopics=[], initializedTopics=[TopicPartitionsInfo(topicId=wO0yz3DfTaCC9fnfn8maXw, topicName='t1', partitions=[0, 1])], deletingTopics=[]) at version 0))]>
at app//org.junit.jupiter.api.AssertionFailureBuilder.build(AssertionFailureBuilder.java:151)
at app//org.junit.jupiter.api.AssertionFailureBuilder.buildAndThrow(AssertionFailureBuilder.java:132)
at app//org.junit.jupiter.api.AssertEquals.failNotEqual(AssertEquals.java:197)
at app//org.junit.jupiter.api.AssertEquals.assertEquals(AssertEquals.java:182)
at app//org.junit.jupiter.api.AssertEquals.assertEquals(AssertEquals.java:177)
at app//org.junit.jupiter.api.Assertions.assertEquals(Assertions.java:1145)
at app//org.apache.kafka.coordinator.group.GroupMetadataManagerTest.testShareGroupInitializeSuccess(GroupMetadataManagerTest.java:21717)
at java.base@21.0.4/java.lang.reflect.Method.invoke(Method.java:580)
at java.base@21.0.4/java.util.ArrayList.forEach(ArrayList.java:1596)
at java.base@21.0.4/java.util.ArrayList.forEach(ArrayList.java:1596)
```
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2101615974
##########
core/src/main/scala/kafka/raft/KafkaMetadataLog.scala:
##########
@@ -589,8 +589,9 @@ object KafkaMetadataLog extends Logging {
): 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)
+ // Since MetadataLogConfig validates the log segment size using a minimum allowed value,
+ // we can safely use `internal.segment.bytes` to configure it instead of relying on `segment.bytes`.
+ props.setProperty(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, config.logSegmentBytes.toString)
Review Comment:
> For another, we should move the test-only constructor out of the production code. This will prevent it from being used accidentally in a production environment. This could be considered as a follow-up ticket
We should probably address this in this PR. Otherwise, the `MetadataLogConfig` constructor that takes `internalLogSegmentBytes` will seem a bit strange in the production code.
see https://github.com/apache/kafka/blob/trunk/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java#L285 as example
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2101057057
##########
core/src/main/scala/kafka/raft/KafkaMetadataLog.scala:
##########
@@ -589,8 +589,9 @@ object KafkaMetadataLog extends Logging {
): 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)
+ // Since MetadataLogConfig validates the log segment size using a minimum allowed value,
+ // we can safely use `internal.segment.bytes` to configure it instead of relying on `segment.bytes`.
+ props.setProperty(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, config.logSegmentBytes.toString)
Review Comment:
Hmm, we have tests using the following constructor to pass in a small segmentBytes value.
```
public MetadataLogConfig(int internalLogSegmentBytes,
long logSegmentMillis,
long retentionMaxBytes,
long retentionMillis,
int maxBatchSizeInBytes,
int maxFetchSizeInBytes,
long deleteDelayMillis) {
```
This will fail the validation in the following if it's passed in as TopicConfig.SEGMENT_BYTES_CONFIG, right?
```
LogConfig.validate(props)
val defaultLogConfig = new LogConfig(props)
```
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2101594593
##########
core/src/main/scala/kafka/raft/KafkaMetadataLog.scala:
##########
@@ -589,8 +589,9 @@ object KafkaMetadataLog extends Logging {
): 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)
+ // Since MetadataLogConfig validates the log segment size using a minimum allowed value,
+ // we can safely use `internal.segment.bytes` to configure it instead of relying on `segment.bytes`.
+ props.setProperty(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, config.logSegmentBytes.toString)
Review Comment:
@junrao You're right. To support smaller segments for Kraft topics, we could reintroduce the broker-level `internal.log.segment.bytes` configuration. However, this duplicates the functionality of the topic-level `internal.segment.bytes` and would affect all topics on the server.
Perhaps we should revisit `internal.metadata.log.segment.bytes` instead. The regular topics could utilize topic-level `internal.segment.bytes` for smaller segments, while Kraft topics would use the broker-level `internal.metadata.log.segment.bytes`. This approach would require adding if-else logic when converting Kraft configurations to log configurations.
```scala
props.setProperty(if config.useInternal LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG else LogConfig.SEGMENT_BYTES_CONFIG, config.logSegmentBytes.toString)
```
For another, we should move the test-only constructor out of the production code. This will prevent it from being used accidentally in a production environment. This could be considered as a follow-up ticket
```java
public MetadataLogConfig(int internalLogSegmentBytes,
long logSegmentMillis,
long retentionMaxBytes,
long retentionMillis,
int maxBatchSizeInBytes,
int maxFetchSizeInBytes,
long deleteDelayMillis) {
```
--
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
Re: [PR] [WIP]KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19371:
URL: https://github.com/apache/kafka/pull/19371#issuecomment-2792577675
for example:
```java
this.internalSegmentSize = getString(TopicConfig.INTERNAL_SEGMENT_BYTES_CONFIG);
...
public long segmentSize() {
if (internalSegmentSize != null) return Long.parseLong(internalSegmentSize);
return segmentMs;
}
```
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2046016358
##########
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:
I am just saying that we now have two different ways to achieve the same goal. In the metadata log approach, you set the desired value through the original config, which is segment.bytes. You then set an internal config to change the min constraint.
The approach in this PR is to set the desired value through a different internal config.
It would be useful to choose same approach for both the metadata log and the regular log.
--
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
Re: [PR] [WIP]KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2073985790
##########
core/src/test/scala/unit/kafka/server/ControllerConfigurationValidatorTest.scala:
##########
@@ -73,15 +74,15 @@ class ControllerConfigurationValidatorTest {
def testValidTopicConfig(): Unit = {
val config = new util.TreeMap[String, String]()
config.put(SEGMENT_JITTER_MS_CONFIG, "1000")
- config.put(SEGMENT_BYTES_CONFIG, "67108864")
+ config.put(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, "67108864")
Review Comment:
This change is unnecessary, right? Ditto below.
##########
core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala:
##########
@@ -1103,7 +1090,7 @@ object KafkaMetadataLogTest {
UnifiedLog.logDirName(KafkaRaftServer.MetadataPartition)
)
- val metadataLog = KafkaMetadataLog(
+ val metadataLog = KafkaMetadataLog.apply(
Review Comment:
Is this change needed?
##########
core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala:
##########
@@ -101,27 +101,27 @@ class DynamicConfigChangeTest extends KafkaServerTestHarness {
val tp = new TopicPartition("test", 0)
val oldSegmentSize = 1000
val logProps = new Properties()
- logProps.put(TopicConfig.SEGMENT_BYTES_CONFIG, oldSegmentSize.toString)
+ logProps.put(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, oldSegmentSize.toString)
Review Comment:
Is this change needed since we don't need a small segment size? Ditto below.
##########
streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java:
##########
@@ -248,7 +247,7 @@ public void consumerConfigMustContainStreamPartitionAssignorConfig() {
props.put(StreamsConfig.PROBING_REBALANCE_INTERVAL_MS_CONFIG, 99_999L);
props.put(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG, 7L);
props.put(StreamsConfig.APPLICATION_SERVER_CONFIG, "dummy:host");
- props.put(StreamsConfig.topicPrefix(TopicConfig.SEGMENT_BYTES_CONFIG), 100);
+ props.put(StreamsConfig.topicPrefix("internal.segment.bytes"), 100);
Review Comment:
Could we use the constant val ? Ditto below.
##########
metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java:
##########
@@ -905,7 +904,7 @@ public void testCreateTopicsWithPolicy() {
null, Map.of()),
new CreateTopicPolicy.RequestMetadata("baz", null, null,
Map.of(0, List.of(2, 1, 0)),
- Map.of(SEGMENT_BYTES_CONFIG, "12300000")),
+ Map.of("internal.segment.bytes", "12300000")),
Review Comment:
Is this change needed? Ditto below.
##########
share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorServiceTest.java:
##########
@@ -1999,7 +1999,7 @@ public void testShareStateTopicConfigs() {
List<String> propNames = List.of(
TopicConfig.CLEANUP_POLICY_CONFIG,
TopicConfig.COMPRESSION_TYPE_CONFIG,
- TopicConfig.SEGMENT_BYTES_CONFIG,
+ "internal.segment.bytes",
Review Comment:
Hmm, this doesn't look right. service.shareGroupStateTopicConfigs() doesn't include this internal config.
##########
raft/src/main/java/org/apache/kafka/raft/MetadataLogConfig.java:
##########
@@ -112,15 +110,14 @@ public class MetadataLogConfig {
* @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.internalLogSegmentBytes = logSegmentBytes;
Review Comment:
Could we add a comment that this constructor is for test only? Could we also rename logSegmentBytes to internalLogSegmentBytes?
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/PurgeRepartitionTopicIntegrationTest.java:
##########
@@ -114,7 +114,7 @@ public final boolean conditionMet() {
.get();
return config.get(TopicConfig.CLEANUP_POLICY_CONFIG).value().equals(TopicConfig.CLEANUP_POLICY_DELETE)
&& config.get(TopicConfig.SEGMENT_MS_CONFIG).value().equals(PURGE_INTERVAL_MS.toString())
- && config.get(TopicConfig.SEGMENT_BYTES_CONFIG).value().equals(PURGE_SEGMENT_BYTES.toString());
+ && config.get("internal.segment.bytes").value().equals(PURGE_SEGMENT_BYTES.toString());
Review Comment:
Could we use the constant val ? Ditto below.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2094016478
##########
core/src/test/scala/unit/kafka/log/LogConfigTest.scala:
##########
@@ -52,8 +52,13 @@ class LogConfigTest {
assertTrue(LogConfig.configNames.asScala
.filter(config => !LogConfig.CONFIGS_WITH_NO_SERVER_DEFAULTS.contains(config))
.forall { config =>
- val serverConfigOpt = LogConfig.serverConfigName(config)
- serverConfigOpt.isPresent && (serverConfigOpt.get != null)
+ // this internal config naming pattern is not as same as a default pattern
+ if (config.equals(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG)) {
Review Comment:
Thanks for information, I think we can remove this test .
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2100943381
##########
core/src/main/scala/kafka/raft/KafkaMetadataLog.scala:
##########
@@ -589,8 +589,9 @@ object KafkaMetadataLog extends Logging {
): 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)
+ // Since MetadataLogConfig validates the log segment size using a minimum allowed value,
+ // we can safely use `internal.segment.bytes` to configure it instead of relying on `segment.bytes`.
+ props.setProperty(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, config.logSegmentBytes.toString)
Review Comment:
> This will fail LogConfig.validate(props) if a test passes in a small log segment value, right?
This won't happen because this PR removes support for configuring small segments for Kraft topics. The minimum Kraft topic size is now 8MB, which is large enough to pass LogConfig.validate(props).
We've removed the configuration directly to streamline the PR, as there are no existing tests in code base for small segment Kraft topics.
If configuring small segments for kraft topic becomes necessary, we will need to reintroduce broker-level internal log configuration, as topic-level configuration is not available for kraft topic.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19371:
URL: https://github.com/apache/kafka/pull/19371#issuecomment-2907762877
@m1a2st could you please tweak your PR description to summarize the changes?
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2098206077
##########
server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java:
##########
@@ -47,9 +48,11 @@ public final class ServerTopicConfigSynonyms {
* both the first and the second synonyms are configured, we will use only the value of
* the first synonym and ignore the second.
*/
- // Topic configs with no mapping to a server config can be found in `LogConfig.CONFIGS_WITH_NO_SERVER_DEFAULTS`
public static final Map<String, List<ConfigSynonym>> ALL_TOPIC_CONFIG_SYNONYMS = Utils.mkMap(
sameNameWithLogPrefix(TopicConfig.SEGMENT_BYTES_CONFIG),
+ // Due to the internal.segment.bytes is in storage module, thus we could not use the
+ // LogConfig#INTERNAL_SEGMENT_BYTES_CONFIG directly. We need to use the string value instead.
+ sameNameWithInternalLogPrefix("internal.segment.bytes"),
Review Comment:
Yes, you are 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2049095738
##########
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:
Personally, I prefer the approach of adding "internal".xxx config as it provide better user experience for public configs, allowing users to see the "correct" min value. Additionally, we can remove the customized logic of validation.
In short, I suggest to add following changes to this PR.
1. remove `METADATA_LOG_SEGMENT_MIN_BYTES_CONFIG`
2. remove `MetadataLogConfig#logSegmentMinBytes`
3. add `internal.metadata.log.segment.bytes`
4. customize `MetadataLogConfig#logSegmentBytes` as following code
```java
public int logSegmentBytes() {
if (internalSogSegmentBytes != null) return internalSogSegmentBytes;
return logSegmentBytes;
}
```
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2042115495
##########
clients/src/main/java/org/apache/kafka/common/config/TopicConfig.java:
##########
@@ -214,6 +214,9 @@ public class TopicConfig {
"configuration. If message.timestamp.type=CreateTime, the message will be rejected if the difference in " +
"timestamps exceeds this specified threshold. This configuration is ignored if message.timestamp.type=LogAppendTime.";
+ // visible for testing
+ public static final String INTERNAL_SEGMENT_BYTES_CONFIG = "internal.segment.bytes";
Review Comment:
`TopicConfig` is a public class, so could you please move it to `LogConfig`?
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2097986937
##########
server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java:
##########
@@ -47,9 +48,11 @@ public final class ServerTopicConfigSynonyms {
* both the first and the second synonyms are configured, we will use only the value of
* the first synonym and ignore the second.
*/
- // Topic configs with no mapping to a server config can be found in `LogConfig.CONFIGS_WITH_NO_SERVER_DEFAULTS`
public static final Map<String, List<ConfigSynonym>> ALL_TOPIC_CONFIG_SYNONYMS = Utils.mkMap(
sameNameWithLogPrefix(TopicConfig.SEGMENT_BYTES_CONFIG),
+ // Due to the internal.segment.bytes is in storage module, thus we could not use the
+ // LogConfig#INTERNAL_SEGMENT_BYTES_CONFIG directly. We need to use the string value instead.
+ sameNameWithInternalLogPrefix("internal.segment.bytes"),
Review Comment:
@m1a2st
Do you mean this PR supports only topic-level internal config? If so, we can't use small segment for Kraft topic as it takes only server-level config. Do I understand correctly?
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2093848474
##########
core/src/test/scala/unit/kafka/log/LogConfigTest.scala:
##########
@@ -52,8 +52,13 @@ class LogConfigTest {
assertTrue(LogConfig.configNames.asScala
.filter(config => !LogConfig.CONFIGS_WITH_NO_SERVER_DEFAULTS.contains(config))
.forall { config =>
- val serverConfigOpt = LogConfig.serverConfigName(config)
- serverConfigOpt.isPresent && (serverConfigOpt.get != null)
+ // this internal config naming pattern is not as same as a default pattern
+ if (config.equals(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG)) {
Review Comment:
I guess this change is related to the comment (https://github.com/apache/kafka/pull/19371#discussion_r2091933863)
Maybe we can remove the test `ensureNoStaticInitializationOrderDependency` and `CONFIGS_WITH_NO_SERVER_DEFAULTS`.
They were used to test the bug caused by the initialization order between `KafkaConfig` and `LogConfig` (see https://github.com/apache/kafka/commit/0ea540552a52b3265c48aacbb9790ea6e71431a5).
Fortunately, the dependence was gone when rewriting `LogConfig` by java (see https://github.com/apache/kafka/commit/96d9710c17b34d4e80259f09845d21de66a5efaf) :)
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2100793274
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java:
##########
@@ -446,6 +447,14 @@ public static List<String> configNames() {
return CONFIG.names().stream().sorted().toList();
}
+ public static List<String> excludeInternalConfigNames() {
Review Comment:
excludeInternalConfigNames => nonInternalConfigNames ?
##########
core/src/main/scala/kafka/raft/KafkaMetadataLog.scala:
##########
@@ -589,8 +589,9 @@ object KafkaMetadataLog extends Logging {
): 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)
+ // Since MetadataLogConfig validates the log segment size using a minimum allowed value,
+ // we can safely use `internal.segment.bytes` to configure it instead of relying on `segment.bytes`.
+ props.setProperty(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, config.logSegmentBytes.toString)
Review Comment:
This could be confusing. If someone turns on the debug logging to see the config for the metadata topic, it will be surprising to see INTERNAL_SEGMENT_BYTES_CONFIG being set, even though it's not set by the users.
--
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
Re: [PR] [WIP]KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #19371:
URL: https://github.com/apache/kafka/pull/19371#issuecomment-2778790334
Hello @junrao, @chia7712
I have a question about this problem.
In [[KAFKA-16368](https://github.com/apache/kafka/pull/18140/files)](https://github.com/apache/kafka/pull/18140/files), only the `ServerLogConfigs` value for `log.segment.bytes` was modified to 1 MB. However, the `TopicConfig` default for `segment.bytes` remains 14 bytes. Many tests rely on this small `segment.bytes` value to generate a large number of segments.
I think there are two possible approaches to resolve this:
1. Add `atLeast(1024 * 1024)` validation in `LogConfig`, which would require fixing around 400 tests.
2. Add a validator in `LogManager#updateTopicConfig` to validate the change request at runtime.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2056483125
##########
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:
@junrao I have discussed with @m1a2st offline, and he will update the PR tomorrow
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2061411092
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -567,7 +568,7 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
client.createAcls(List(denyAcl).asJava, new CreateAclsOptions()).all().get()
val topics = Seq(topic1, topic2)
- val configsOverride = Map(TopicConfig.SEGMENT_BYTES_CONFIG -> "100000").asJava
+ val configsOverride = Map(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG -> "100000").asJava
Review Comment:
this test case is used to verify the custom topic-level config, so we can increase the value to make test pass.
##########
raft/src/main/java/org/apache/kafka/raft/MetadataLogConfig.java:
##########
@@ -52,14 +51,14 @@ public class MetadataLogConfig {
"configuration. The Kafka node will generate a snapshot when either the maximum time interval is reached or the " +
"maximum bytes limit is reached.";
- public static final String METADATA_LOG_SEGMENT_MIN_BYTES_CONFIG = "metadata.log.segment.min.bytes";
- public static final String METADATA_LOG_SEGMENT_MIN_BYTES_DOC = "Override the minimum size for a single metadata log file. This should be used for testing only.";
- public static final int METADATA_LOG_SEGMENT_MIN_BYTES_DEFAULT = 8 * 1024 * 1024;
-
public static final String METADATA_LOG_SEGMENT_BYTES_CONFIG = "metadata.log.segment.bytes";
public static final String METADATA_LOG_SEGMENT_BYTES_DOC = "The maximum size of a single metadata log file.";
public static final int METADATA_LOG_SEGMENT_BYTES_DEFAULT = 1024 * 1024 * 1024;
+ public static final String INTERNAL_METADATA_LOG_SEGMENT_BYTES_CONFIG = "internal.metadata.log.segment.bytes";
Review Comment:
Do we need another internal config? User can configure `internal.segment.bytes` if they want to create small segment, right?
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -601,7 +602,7 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
assertNotEquals(Uuid.ZERO_UUID, createResult.topicId(topic1).get())
assertEquals(topicIds(topic1), createResult.topicId(topic1).get())
assertFutureThrows(classOf[TopicAuthorizationException], createResult.topicId(topic2))
-
+
Review Comment:
please revert unrelated change
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2093536869
##########
core/src/test/scala/unit/kafka/log/LogConfigTest.scala:
##########
@@ -52,8 +52,13 @@ class LogConfigTest {
assertTrue(LogConfig.configNames.asScala
.filter(config => !LogConfig.CONFIGS_WITH_NO_SERVER_DEFAULTS.contains(config))
.forall { config =>
- val serverConfigOpt = LogConfig.serverConfigName(config)
- serverConfigOpt.isPresent && (serverConfigOpt.get != null)
+ // this internal config naming pattern is not as same as a default pattern
+ if (config.equals(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG)) {
Review Comment:
Why do we need this? ALL_TOPIC_CONFIG_SYNONYMS includes INTERNAL_SEGMENT_BYTES_CONFIG.
##########
server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java:
##########
@@ -50,6 +50,7 @@ public final class ServerTopicConfigSynonyms {
// Topic configs with no mapping to a server config can be found in `LogConfig.CONFIGS_WITH_NO_SERVER_DEFAULTS`
public static final Map<String, List<ConfigSynonym>> ALL_TOPIC_CONFIG_SYNONYMS = Utils.mkMap(
sameNameWithLogPrefix(TopicConfig.SEGMENT_BYTES_CONFIG),
+ sameName("internal.log." + TopicConfig.SEGMENT_BYTES_CONFIG),
Review Comment:
This is not really the same name. Could we add a method like sameNameWithInternalLogPrefix?
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19371:
URL: https://github.com/apache/kafka/pull/19371#issuecomment-2907811568
@junrao @m1a2st Thanks for all your patient reviews and updates. I'm going to merge this. Please feel free to leave any further comments for follow-up.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2102394876
##########
core/src/main/scala/kafka/raft/KafkaMetadataLog.scala:
##########
@@ -589,8 +589,9 @@ object KafkaMetadataLog extends Logging {
): 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)
+ // Since MetadataLogConfig validates the log segment size using a minimum allowed value,
+ // we can safely use `internal.segment.bytes` to configure it instead of relying on `segment.bytes`.
+ props.setProperty(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, config.logSegmentBytes.toString)
Review Comment:
I believe we shouldn’t move this constructor outside of `MetadataLogConfig`, because the public `MetadataLogConfig(AbstractConfig config)` constructor doesn’t allow us to set properties like `maxBatchSizeInBytes`, `maxFetchSizeInBytes`, or `deleteDelayMillis`.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2045952460
##########
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:
Excuse me, the strategy used by metadata log is to add a "internal" config (`METADATA_LOG_SEGMENT_MIN_BYTES_CONFIG`) to change the (metadata) segment size in testing, and that is what we want to address in this PR - we add a "internal" config for regular log, and so the test can use the "smaller" segment 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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2045151038
##########
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:
+1 to move the internal config to `MetadataLogConfig`, and it would be better to wait #19465 extracting the metadata-related configs from other class to `MetadataLogConfig`
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on PR #19371:
URL: https://github.com/apache/kafka/pull/19371#issuecomment-2887628228
It's useful to check if metadata.log.segment.min.bytes is used in any ducktape tests.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2086829441
##########
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:
I think we shouldn’t add this unless it’s actually needed by the test. Given that it’s an internal configuration, it might be more appropriate to introduce it only when a concrete use case arises.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19371:
URL: https://github.com/apache/kafka/pull/19371#issuecomment-2898754119
@m1a2st please fix the conflicts
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2103037928
##########
raft/src/main/java/org/apache/kafka/raft/MetadataLogConfig.java:
##########
@@ -80,72 +79,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, null, null, LOW, INTERNAL_MAX_BATCH_SIZE_IN_BYTES_DOC)
+ .defineInternal(INTERNAL_MAX_FETCH_SIZE_IN_BYTES_CONFIG, INT, null, null, LOW, INTERNAL_MAX_FETCH_SIZE_IN_BYTES_DOC)
+ .defineInternal(INTERNAL_DELETE_DELAY_MILLIS_CONFIG, LONG, null, 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;
- }
-
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.maxBatchSizeInBytes = Objects.requireNonNullElse(config.getInt(INTERNAL_MAX_BATCH_SIZE_IN_BYTES_CONFIG), KafkaRaftClient.MAX_BATCH_SIZE_BYTES);
Review Comment:
Should we name `maxBatchSizeInBytes` and the other two with the internal prefix? Also, should we just use KafkaRaftClient.MAX_BATCH_SIZE_BYTES as the default in the config def?
##########
raft/src/main/java/org/apache/kafka/raft/MetadataLogConfig.java:
##########
@@ -80,72 +79,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, null, null, LOW, INTERNAL_MAX_BATCH_SIZE_IN_BYTES_DOC)
+ .defineInternal(INTERNAL_MAX_FETCH_SIZE_IN_BYTES_CONFIG, INT, null, null, LOW, INTERNAL_MAX_FETCH_SIZE_IN_BYTES_DOC)
+ .defineInternal(INTERNAL_DELETE_DELAY_MILLIS_CONFIG, LONG, null, 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;
- }
-
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.maxBatchSizeInBytes = Objects.requireNonNullElse(config.getInt(INTERNAL_MAX_BATCH_SIZE_IN_BYTES_CONFIG), KafkaRaftClient.MAX_BATCH_SIZE_BYTES);
+ this.maxFetchSizeInBytes = Objects.requireNonNullElse(config.getInt(INTERNAL_MAX_FETCH_SIZE_IN_BYTES_CONFIG), KafkaRaftClient.MAX_FETCH_SIZE_BYTES);
+ this.deleteDelayMillis = Objects.requireNonNullElse(config.getLong(INTERNAL_DELETE_DELAY_MILLIS_CONFIG), ServerLogConfigs.LOG_DELETE_DELAY_MS_DEFAULT);
}
public int logSegmentBytes() {
- return logSegmentBytes;
+ return Objects.requireNonNullElse(internalSegmentBytes, logSegmentBytes);
Review Comment:
Hmm, it seems it's more intuitive to just return `logSegmentBytes` since we have a separate public api to return `internalSegmentBytes`.
##########
core/src/main/scala/kafka/raft/KafkaMetadataLog.scala:
##########
@@ -589,8 +589,10 @@ object KafkaMetadataLog extends Logging {
): 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)
+ if (config.internalSegmentBytes() != null)
+ props.setProperty(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, config.logSegmentBytes.toString)
Review Comment:
Hmm, `config.logSegmentBytes` should be `config.internalLogSegmentBytes`, right?
##########
core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala:
##########
@@ -1166,4 +1161,25 @@ object KafkaMetadataLogTest {
}
dir
}
+
+ private def createMetadataLogConfig(
+ internalLogSegmentBytes: Int,
+ logSegmentMillis: Long,
+ retentionMaxBytes: Long,
+ retentionMillis: Long,
+ maxBatchSizeInBytes: Int,
Review Comment:
Should these three be named internal? Ditto in other test.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2102447675
##########
core/src/main/scala/kafka/raft/KafkaMetadataLog.scala:
##########
@@ -589,8 +589,9 @@ object KafkaMetadataLog extends Logging {
): 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)
+ // Since MetadataLogConfig validates the log segment size using a minimum allowed value,
+ // we can safely use `internal.segment.bytes` to configure it instead of relying on `segment.bytes`.
+ props.setProperty(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, config.logSegmentBytes.toString)
Review Comment:
> maxBatchSizeInBytes, maxFetchSizeInBytes, or deleteDelayMillis.
Maybe it is time to introduce internal configs for them? That make `MetadataLogConfig` consistent with other config classes which allow to configure all variables.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19371:
URL: https://github.com/apache/kafka/pull/19371#issuecomment-2913172227
@mumrah thanks for your reviews. We will file a minor later
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2112236030
##########
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:
Since `config.getInt(INTERNAL_METADATA_LOG_SEGMENT_BYTES_CONFIG)` may return a null value, using `OptionalInt` is not suitable here.
If we want to wrap the result with `Optional`, I think `Optional<Integer>` would be more appropriate.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2103145357
##########
raft/src/main/java/org/apache/kafka/raft/MetadataLogConfig.java:
##########
@@ -80,72 +79,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, null, null, LOW, INTERNAL_MAX_BATCH_SIZE_IN_BYTES_DOC)
+ .defineInternal(INTERNAL_MAX_FETCH_SIZE_IN_BYTES_CONFIG, INT, null, null, LOW, INTERNAL_MAX_FETCH_SIZE_IN_BYTES_DOC)
+ .defineInternal(INTERNAL_DELETE_DELAY_MILLIS_CONFIG, LONG, null, 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;
- }
-
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.maxBatchSizeInBytes = Objects.requireNonNullElse(config.getInt(INTERNAL_MAX_BATCH_SIZE_IN_BYTES_CONFIG), KafkaRaftClient.MAX_BATCH_SIZE_BYTES);
+ this.maxFetchSizeInBytes = Objects.requireNonNullElse(config.getInt(INTERNAL_MAX_FETCH_SIZE_IN_BYTES_CONFIG), KafkaRaftClient.MAX_FETCH_SIZE_BYTES);
+ this.deleteDelayMillis = Objects.requireNonNullElse(config.getLong(INTERNAL_DELETE_DELAY_MILLIS_CONFIG), ServerLogConfigs.LOG_DELETE_DELAY_MS_DEFAULT);
}
public int logSegmentBytes() {
- return logSegmentBytes;
+ return Objects.requireNonNullElse(internalSegmentBytes, logSegmentBytes);
Review Comment:
I think there are two styles:
```java
public int logSegmentBytes() {
return Objects.requireNonNullElse(internalSegmentBytes, logSegmentBytes);
}
public boolean useInternalSegmentBytes() {
return internalSegmentBytes != null;
}
if (config.useInternalSegmentBytes())
props.setProperty(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, config.logSegmentBytes.toString)
else
props.setProperty(TopicConfig.SEGMENT_BYTES_CONFIG, config.logSegmentBytes.toString)
```
```java
public int logSegmentBytes() {
return logSegmentBytes;
}
public Integer internalSegmentBytes() {
return internalSegmentBytes;
}
if (config.internalSegmentBytes() != null)
props.setProperty(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, config.internalSegmentBytes.toString)
else
props.setProperty(TopicConfig.SEGMENT_BYTES_CONFIG, config.logSegmentBytes.toString)
```
I prefer the later since `useInternalSegmentBytes` is not usual in config classes
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2097427388
##########
server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java:
##########
@@ -47,9 +48,11 @@ public final class ServerTopicConfigSynonyms {
* both the first and the second synonyms are configured, we will use only the value of
* the first synonym and ignore the second.
*/
- // Topic configs with no mapping to a server config can be found in `LogConfig.CONFIGS_WITH_NO_SERVER_DEFAULTS`
public static final Map<String, List<ConfigSynonym>> ALL_TOPIC_CONFIG_SYNONYMS = Utils.mkMap(
sameNameWithLogPrefix(TopicConfig.SEGMENT_BYTES_CONFIG),
+ // Due to the internal.segment.bytes is in storage module, thus we could not use the
+ // LogConfig#INTERNAL_SEGMENT_BYTES_CONFIG directly. We need to use the string value instead.
+ sameNameWithInternalLogPrefix("internal.segment.bytes"),
Review Comment:
We currently don't have small segment tests for Kraft topics. Therefore, a more effective way to streamline this PR would be to support topic-level configuration only. This means the new internal configuration (`internal.segment.bytes`) would apply exclusively to user topics.
We can revisit this config for Kraft topics in the future if it needs.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #19371:
URL: https://github.com/apache/kafka/pull/19371
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #19371:
URL: https://github.com/apache/kafka/pull/19371#issuecomment-2907775441
Thanks for reminder, updated it.
--
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
Re: [PR] [WIP]KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on PR #19371:
URL: https://github.com/apache/kafka/pull/19371#issuecomment-2787910437
@m1a2st : Perhaps we could somehow allow the tests to set a small `segment.bytes`.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2043103248
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java:
##########
@@ -186,12 +188,13 @@ public Optional<String> serverConfigName(String configName) {
.define(ServerLogConfigs.CREATE_TOPIC_POLICY_CLASS_NAME_CONFIG, CLASS, null, LOW, ServerLogConfigs.CREATE_TOPIC_POLICY_CLASS_NAME_DOC)
.define(ServerLogConfigs.ALTER_CONFIG_POLICY_CLASS_NAME_CONFIG, CLASS, null, LOW, ServerLogConfigs.ALTER_CONFIG_POLICY_CLASS_NAME_DOC)
.define(ServerLogConfigs.LOG_DIR_FAILURE_TIMEOUT_MS_CONFIG, LONG, ServerLogConfigs.LOG_DIR_FAILURE_TIMEOUT_MS_DEFAULT, atLeast(1), LOW, ServerLogConfigs.LOG_DIR_FAILURE_TIMEOUT_MS_DOC)
- .defineInternal(ServerLogConfigs.LOG_INITIAL_TASK_DELAY_MS_CONFIG, LONG, ServerLogConfigs.LOG_INITIAL_TASK_DELAY_MS_DEFAULT, atLeast(0), LOW, ServerLogConfigs.LOG_INITIAL_TASK_DELAY_MS_DOC);
+ .defineInternal(ServerLogConfigs.LOG_INITIAL_TASK_DELAY_MS_CONFIG, LONG, ServerLogConfigs.LOG_INITIAL_TASK_DELAY_MS_DEFAULT, atLeast(0), LOW, ServerLogConfigs.LOG_INITIAL_TASK_DELAY_MS_DOC)
+ .defineInternal(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, INT, null, null, LOW, ServerLogConfigs.LOG_SEGMENT_BYTES_DOC);
Review Comment:
Could we document that this config is for testing?
##########
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:
Instead of having an `internalApply`, could we just use the existing `apply` and add `INTERNAL_SEGMENT_BYTES_CONFIG` to `MetadataLogConfig`?
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2106061094
##########
core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala:
##########
@@ -1166,4 +1161,25 @@ object KafkaMetadataLogTest {
}
dir
}
+
+ private def createMetadataLogConfig(
+ internalLogSegmentBytes: Int,
+ logSegmentMillis: Long,
+ retentionMaxBytes: Long,
+ retentionMillis: Long,
+ internalMaxBatchSizeInBytes: Int,
+ internalMaxFetchSizeInBytes: Int,
+ internalDeleteDelayMillis: Long
+ ): MetadataLogConfig = {
+ val config: util.Map[String, Any] = util.Map.of(
Review Comment:
We need to explicitly declare entries as `Any`, otherwise the compiler will infer it as `long`, which would cause errors when parsing the config.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #19371:
URL: https://github.com/apache/kafka/pull/19371#issuecomment-2798418146
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
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
Re: [PR] [WIP]KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19371:
URL: https://github.com/apache/kafka/pull/19371#issuecomment-2790393621
Maybe we can add a internal config to allow tests to define small 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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2045280340
##########
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:
I just realized that metadata log uses a different approach to allow tests to use a smaller segment bytes than allowed in production. That approach defines the original segment byte config with a small minimal requirement, but adds METADATA_LOG_SEGMENT_MIN_BYTES_CONFIG to enforce the actual minimal requirement in production. This new config could be changed in tests to allow for smaller minimal bytes. The benefit of this approach is that it allows the existing config to be used directly to set a smaller value for tests. The downside is that the doc for min value is inaccurate and the validation is done through a customized logic.
It would be useful to pick the same strategy between metadata log and regular log. The metadata log approach seems slightly better since it's less intrusive. We could fix the inaccurate min value description for production somehow.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2084656210
##########
streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java:
##########
@@ -248,7 +247,7 @@ public void consumerConfigMustContainStreamPartitionAssignorConfig() {
props.put(StreamsConfig.PROBING_REBALANCE_INTERVAL_MS_CONFIG, 99_999L);
props.put(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG, 7L);
props.put(StreamsConfig.APPLICATION_SERVER_CONFIG, "dummy:host");
- props.put(StreamsConfig.topicPrefix(TopicConfig.SEGMENT_BYTES_CONFIG), 100);
+ props.put(StreamsConfig.topicPrefix("internal.segment.bytes"), 100);
Review Comment:
We can't add the storage dependency to the Streams module because the Streams module targets Java 11, while the Storage module requires Java 17.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #19371:
URL: https://github.com/apache/kafka/pull/19371#issuecomment-2917938422
I’ve addressed the follow-up comments on https://github.com/apache/kafka/pull/19842
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19371:
URL: https://github.com/apache/kafka/pull/19371#issuecomment-2806799851
@m1a2st could you please check the failed tests?
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2100899250
##########
core/src/main/scala/kafka/raft/KafkaMetadataLog.scala:
##########
@@ -589,8 +589,9 @@ object KafkaMetadataLog extends Logging {
): 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)
+ // Since MetadataLogConfig validates the log segment size using a minimum allowed value,
+ // we can safely use `internal.segment.bytes` to configure it instead of relying on `segment.bytes`.
+ props.setProperty(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, config.logSegmentBytes.toString)
Review Comment:
Hmm, does that work? This will fail `LogConfig.validate(props)` if a test passes in a small log segment value, 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2100843498
##########
core/src/main/scala/kafka/raft/KafkaMetadataLog.scala:
##########
@@ -589,8 +589,9 @@ object KafkaMetadataLog extends Logging {
): 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)
+ // Since MetadataLogConfig validates the log segment size using a minimum allowed value,
+ // we can safely use `internal.segment.bytes` to configure it instead of relying on `segment.bytes`.
+ props.setProperty(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, config.logSegmentBytes.toString)
Review Comment:
We don't need this change now, as the internal config for kraft topic is removed. Additionally, the min size of kraft topic segment (8MB) is larger than size of regular segment (1MB). Hence, it is pretty safe to set `props.setProperty(TopicConfig.SEGMENT_MS_CONFIG, config.logSegmentMillis.toString)`
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2097948691
##########
server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java:
##########
@@ -47,9 +48,11 @@ public final class ServerTopicConfigSynonyms {
* both the first and the second synonyms are configured, we will use only the value of
* the first synonym and ignore the second.
*/
- // Topic configs with no mapping to a server config can be found in `LogConfig.CONFIGS_WITH_NO_SERVER_DEFAULTS`
public static final Map<String, List<ConfigSynonym>> ALL_TOPIC_CONFIG_SYNONYMS = Utils.mkMap(
sameNameWithLogPrefix(TopicConfig.SEGMENT_BYTES_CONFIG),
+ // Due to the internal.segment.bytes is in storage module, thus we could not use the
+ // LogConfig#INTERNAL_SEGMENT_BYTES_CONFIG directly. We need to use the string value instead.
+ sameNameWithInternalLogPrefix("internal.segment.bytes"),
Review Comment:
In the previous solution, we required the server-level `internal.log.segment.bytes` to configure `MetadataLogConfig`.
However, since we now always use `internal.segment.bytes` in the `KafkaMetadataLog.apply` method, this server-level config is no longer necessary and can be removed.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #19371:
URL: https://github.com/apache/kafka/pull/19371#issuecomment-2905917690
I merge trunk, and retrigger the CI again.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #19371:
URL: https://github.com/apache/kafka/pull/19371#issuecomment-2872391463
Hello @chia7712, @junrao, Sorry for the late reply.
I’ve updated the patch. Developers can still use the internal config at both the server and topic levels. I also updated the logic so that if a user configures the topic-level config without specifying the internal one, we will no longer return the internal config to the user.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2051911610
##########
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:
Thanks for @chia7712, @junrao comments, addressed it :)
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #19371:
URL: https://github.com/apache/kafka/pull/19371#issuecomment-2890936867
> It's useful to check if metadata.log.segment.min.bytes is used in any ducktape tests.
I’ve checked all the tests in the Kafka project, and none of them use this configuration.
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2096192551
##########
server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java:
##########
@@ -47,9 +48,11 @@ public final class ServerTopicConfigSynonyms {
* both the first and the second synonyms are configured, we will use only the value of
* the first synonym and ignore the second.
*/
- // Topic configs with no mapping to a server config can be found in `LogConfig.CONFIGS_WITH_NO_SERVER_DEFAULTS`
public static final Map<String, List<ConfigSynonym>> ALL_TOPIC_CONFIG_SYNONYMS = Utils.mkMap(
sameNameWithLogPrefix(TopicConfig.SEGMENT_BYTES_CONFIG),
+ // Due to the internal.segment.bytes is in storage module, thus we could not use the
+ // LogConfig#INTERNAL_SEGMENT_BYTES_CONFIG directly. We need to use the string value instead.
+ sameNameWithInternalLogPrefix("internal.segment.bytes"),
Review Comment:
Is the intention to support `internal.segment.bytes` at the broker level? If so, should we add it to `KafkaConfig.extractLogConfigMap` too?
Currently, if internal.log.segment.bytes is set at the broker, it won't be picked up by a new user topic, but it will be picked up by the kraft topic.
Thinking a bit more. Do we need to support internal.segment.bytes at the broker level? For testing, is it enough to just support internal.segment.bytes at the topic level?
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19371:
URL: https://github.com/apache/kafka/pull/19371#discussion_r2105869770
##########
storage/src/test/java/org/apache/kafka/storage/internals/log/LocalLogTest.java:
##########
@@ -198,7 +197,7 @@ public void testUpdateConfig() {
assertEquals(oldConfig, log.config());
Properties props = new Properties();
- props.put(TopicConfig.SEGMENT_BYTES_CONFIG, oldConfig.segmentSize + 1);
+ props.put(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, oldConfig.segmentSize() + 1);
Review Comment:
it can keep using `TopicConfig.SEGMENT_BYTES_CONFIG`, right?
##########
core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala:
##########
@@ -95,29 +95,29 @@ class DynamicConfigChangeTest extends KafkaServerTestHarness {
@Test
def testDynamicTopicConfigChange(): Unit = {
val tp = new TopicPartition("test", 0)
- val oldSegmentSize = 1000
+ val oldSegmentSize = 2 * 1024 * 1024
val logProps = new Properties()
logProps.put(TopicConfig.SEGMENT_BYTES_CONFIG, oldSegmentSize.toString)
createTopic(tp.topic, 1, 1, logProps)
TestUtils.retry(10000) {
val logOpt = this.brokers.head.logManager.getLog(tp)
assertTrue(logOpt.isDefined)
- assertEquals(oldSegmentSize, logOpt.get.config.segmentSize)
+ assertEquals(oldSegmentSize, logOpt.get.config.segmentSize())
}
val newSegmentSize = 2000
val admin = createAdminClient()
try {
val resource = new ConfigResource(ConfigResource.Type.TOPIC, tp.topic())
- val op = new AlterConfigOp(new ConfigEntry(TopicConfig.SEGMENT_BYTES_CONFIG, newSegmentSize.toString),
+ val op = new AlterConfigOp(new ConfigEntry(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, newSegmentSize.toString),
Review Comment:
this test is used to verify the alter, so you can use larger `newSegmentSize` to keep using `TopicConfig.SEGMENT_BYTES_CONFIG`
##########
streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java:
##########
@@ -248,7 +247,7 @@ public void consumerConfigMustContainStreamPartitionAssignorConfig() {
props.put(StreamsConfig.PROBING_REBALANCE_INTERVAL_MS_CONFIG, 99_999L);
props.put(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG, 7L);
props.put(StreamsConfig.APPLICATION_SERVER_CONFIG, "dummy:host");
- props.put(StreamsConfig.topicPrefix(TopicConfig.SEGMENT_BYTES_CONFIG), 100);
+ props.put(StreamsConfig.topicPrefix("internal.segment.bytes"), 100);
Review Comment:
this is used to test the updated configs get returned, so we can increase the configured value instead of using internal config.
##########
core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala:
##########
@@ -1166,4 +1161,25 @@ object KafkaMetadataLogTest {
}
dir
}
+
+ private def createMetadataLogConfig(
+ internalLogSegmentBytes: Int,
+ logSegmentMillis: Long,
+ retentionMaxBytes: Long,
+ retentionMillis: Long,
+ internalMaxBatchSizeInBytes: Int,
+ internalMaxFetchSizeInBytes: Int,
+ internalDeleteDelayMillis: Long
+ ): MetadataLogConfig = {
+ val config: util.Map[String, Any] = util.Map.of(
Review Comment:
`val config = util.Map.of(`
##########
core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala:
##########
@@ -1166,4 +1161,25 @@ object KafkaMetadataLogTest {
}
dir
}
+
+ private def createMetadataLogConfig(
+ internalLogSegmentBytes: Int,
+ logSegmentMillis: Long,
+ retentionMaxBytes: Long,
+ retentionMillis: Long,
+ internalMaxBatchSizeInBytes: Int,
Review Comment:
Could you please add default value to `internalMaxBatchSizeInBytes`, `internalMaxFetchSizeInBytes`, and `internalDeleteDelayMillis` to streamline the callers?
##########
metadata/src/main/java/org/apache/kafka/metadata/KafkaConfigSchema.java:
##########
@@ -166,6 +166,10 @@ public Map<String, ConfigEntry> resolveEffectiveTopicConfigs(
ConfigDef configDef = configDefs.getOrDefault(ConfigResource.Type.TOPIC, EMPTY_CONFIG_DEF);
HashMap<String, ConfigEntry> effectiveConfigs = new HashMap<>();
for (ConfigDef.ConfigKey configKey : configDef.configKeys().values()) {
+ // This config is internal; if the user hasn't set it explicitly, it should not be returned.
+ if (configKey.internalConfig && !dynamicTopicConfigs.containsKey(configKey.name)) {
Review Comment:
Have you added test for this change?
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java:
##########
@@ -446,6 +447,14 @@ public static List<String> configNames() {
return CONFIG.names().stream().sorted().toList();
}
+ public static List<String> nonInternalConfigNames() {
+ return CONFIG.configKeys().entrySet()
+ .stream()
+ .filter(entry -> !entry.getValue().internalConfig)
+ .map(Map.Entry::getKey)
+ .sorted().toList();
+ }
+
public static Optional<String> serverConfigName(String configName) {
Review Comment:
it is unused now. please remove it
--
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
Re: [PR] KAFKA-19080 The constraint on segment.ms is not enforced at topic level [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #19371:
URL: https://github.com/apache/kafka/pull/19371#issuecomment-2796373720
Thanks for @junrao, @chia7712 comments, This PR is already for review, PTAL.
--
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