[PR] KAFKA-20388: Move LogConfigTest from core to storage [kafka]

["see-quick (via GitHub)" <gi...@apache.org>]

see-quick opened a new pull request, #21980:
URL: https://github.com/apache/kafka/pull/21980

   This is another PR migrating Log* test classes into storage module (from core). 
   I have migrate this but also did a bit refactoring ... where I though it would remove a bit of redudant code (i.e., I have created a helper method for `KafkaConfig`). Also improved a bit `testFromPropsInvalid` where there were two branches pointing on the same values so I merged those. And also I had problem with `testFromPropsInvalid` where it had CyclomaticComplexity so I re-write it into more cleaner approach just extracting the switch into helper method. 


-- 
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-20388: Move LogConfigTest from core to storage [kafka]

["see-quick (via GitHub)" <gi...@apache.org>]

see-quick commented on code in PR #21980:
URL: https://github.com/apache/kafka/pull/21980#discussion_r3073364776


##########
storage/src/test/java/org/apache/kafka/storage/internals/log/LogConfigTest.java:
##########
@@ -0,0 +1,460 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.storage.internals.log;
+
+import kafka.server.KafkaConfig;
+import kafka.utils.TestUtils;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.errors.InvalidConfigurationException;
+import org.apache.kafka.server.config.ServerLogConfigs;
+import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class LogConfigTest {
+
+    @Test
+    public void testKafkaConfigToProps() {
+        long millisInHour = 60L * 60L * 1000L;
+        long millisInDay = 24L * millisInHour;
+        long bytesInGB = 1024L * 1024L * 1024L;
+        Properties kafkaProps = TestUtils.createDummyBrokerConfig();
+        kafkaProps.put(ServerLogConfigs.LOG_ROLL_TIME_HOURS_CONFIG, "2");
+        kafkaProps.put(ServerLogConfigs.LOG_ROLL_TIME_JITTER_HOURS_CONFIG, "2");
+        kafkaProps.put(ServerLogConfigs.LOG_RETENTION_TIME_HOURS_CONFIG, "960"); // 40 days
+        kafkaProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, "2592000000"); // 30 days
+        kafkaProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, "4294967296"); // 4 GB
+
+        Map<String, Object> logProps = KafkaConfig.fromProps(kafkaProps).extractLogConfigMap();
+        assertEquals(2 * millisInHour, logProps.get(TopicConfig.SEGMENT_MS_CONFIG));
+        assertEquals(2 * millisInHour, logProps.get(TopicConfig.SEGMENT_JITTER_MS_CONFIG));
+        assertEquals(40 * millisInDay, logProps.get(TopicConfig.RETENTION_MS_CONFIG));
+        assertEquals(30 * millisInDay, logProps.get(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG));
+        assertEquals(4 * bytesInGB, logProps.get(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG));
+    }
+
+    @Test
+    public void testFromPropsInvalid() {
+        for (String name : LogConfig.configNames()) {
+            Object[] invalidValues = invalidValuesForProp(name);
+            if (invalidValues != null) {
+                assertPropertyInvalid(name, invalidValues);
+            }
+        }
+    }
+
+    private Object[] invalidValuesForProp(String name) {
+        return switch (name) {
+            case TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG -> new Object[]{"not a boolean"};
+            case TopicConfig.RETENTION_BYTES_CONFIG, TopicConfig.RETENTION_MS_CONFIG -> new Object[]{"not_a_number"};
+            case TopicConfig.CLEANUP_POLICY_CONFIG -> new Object[]{"true", "foobar"};
+            case TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG -> new Object[]{"not_a_number", "-0.1", "1.2"};
+            case TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG -> new Object[]{"not_a_number", "0", "-1"};
+            case TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG -> new Object[]{"not_a_number", "-3"};
+            case TopicConfig.COMPRESSION_GZIP_LEVEL_CONFIG -> new Object[]{"not_a_number", "-2"};
+            case TopicConfig.COMPRESSION_LZ4_LEVEL_CONFIG -> new Object[]{"not_a_number", "-1"};
+            case TopicConfig.COMPRESSION_ZSTD_LEVEL_CONFIG -> new Object[]{"not_a_number", "-0.1"};
+            case TopicConfig.REMOTE_LOG_COPY_DISABLE_CONFIG, TopicConfig.REMOTE_LOG_DELETE_ON_DISABLE_CONFIG -> new Object[]{"not_a_number", "remove", "0"};
+            case LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG -> null;
+            default -> new Object[]{"not_a_number", "-1"};
+        };
+    }
+
+    @Test
+    public void testInvalidCompactionLagConfig() {
+        HashMap<String, String> props = new HashMap<>();
+        props.put(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG, "100");
+        props.put(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG, "200");
+        assertThrows(Exception.class, () -> LogConfig.validate(props));
+    }
+
+    @Test
+    public void shouldValidateThrottledReplicasConfig() {
+        assertTrue(isValid("*"));
+        assertTrue(isValid("* "));
+        assertTrue(isValid(""));
+        assertTrue(isValid(" "));
+        assertTrue(isValid("100:10"));
+        assertTrue(isValid("100:10,12:10"));
+        assertTrue(isValid("100:10,12:10,15:1"));
+        assertTrue(isValid("100:10,12:10,15:1  "));
+        assertTrue(isValid("100:0,"));
+
+        assertFalse(isValid("100"));
+        assertFalse(isValid("100:"));
+        assertFalse(isValid("100:0,10"));
+        assertFalse(isValid("100:0,10:"));
+        assertFalse(isValid("100:0,10:   "));
+        assertFalse(isValid("100 :0,10:   "));
+        assertFalse(isValid("100: 0,10:   "));
+        assertFalse(isValid("100:0,10 :   "));
+        assertFalse(isValid("*,100:10"));
+        assertFalse(isValid("* ,100:10"));
+    }
+
+    /* Sanity check that toHtmlTable produces one of the expected configs */
+    @Test
+    public void testToHtmlTable() {
+        String html = LogConfig.configDefCopy().toHtmlTable();
+        String expectedConfig = "<td>file.delete.delay.ms</td>";
+        assertTrue(html.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + html);
+    }
+
+    /* Sanity check that toHtml produces one of the expected configs */
+    @Test
+    public void testToHtml() {
+        String html = LogConfig.configDefCopy().toHtml(4, key -> "prefix_" + key, Map.of());
+        String expectedConfig = "<h4><a id=\"file.delete.delay.ms\"></a><a id=\"prefix_file.delete.delay.ms\" href=\"#prefix_file.delete.delay.ms\">file.delete.delay.ms</a></h4>";
+        assertTrue(html.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + html);
+    }
+
+    /* Sanity check that toEnrichedRst produces one of the expected configs */
+    @Test
+    public void testToEnrichedRst() {
+        String rst = LogConfig.configDefCopy().toEnrichedRst();
+        String expectedConfig = "``file.delete.delay.ms``";
+        assertTrue(rst.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + rst);
+    }
+
+    /* Sanity check that toRst produces one of the expected configs */
+    @Test
+    public void testToRst() {
+        String rst = LogConfig.configDefCopy().toRst();
+        String expectedConfig = "``file.delete.delay.ms``";
+        assertTrue(rst.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + rst);
+    }
+
+    @Test
+    public void testGetConfigValue() {
+        // Add a config that doesn't set the `serverDefaultConfigName`
+        LogConfig.LogConfigDef configDef = LogConfig.configDefCopy();
+        String configNameWithNoServerMapping = "log.foo";
+        configDef.define(configNameWithNoServerMapping, ConfigDef.Type.INT, 1, ConfigDef.Importance.MEDIUM,
+                configNameWithNoServerMapping + " doc");
+
+        ConfigDef.ConfigKey deleteDelayKey = configDef.configKeys().get(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG);
+        Object deleteDelayServerDefault = configDef.getConfigValue(deleteDelayKey, LogConfig.SERVER_DEFAULT_HEADER_NAME);
+        assertEquals(ServerLogConfigs.LOG_DELETE_DELAY_MS_CONFIG, deleteDelayServerDefault);
+
+        ConfigDef.ConfigKey keyWithNoServerMapping = configDef.configKeys().get(configNameWithNoServerMapping);
+        Object nullServerDefault = configDef.getConfigValue(keyWithNoServerMapping, LogConfig.SERVER_DEFAULT_HEADER_NAME);
+        assertNull(nullServerDefault);
+    }
+
+    @Test
+    public void testOverriddenConfigsAsLoggableString() {
+        Properties kafkaProps = TestUtils.createDummyBrokerConfig();
+        kafkaProps.put("unknown.broker.password.config", "aaaaa");
+        kafkaProps.put(ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG, "50");
+        kafkaProps.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, "somekeypassword");
+        KafkaConfig kafkaConfig = KafkaConfig.fromProps(kafkaProps);
+        Properties topicOverrides = new Properties();
+        // Only set as a topic config
+        topicOverrides.setProperty(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2");
+        // Overrides value from broker config
+        topicOverrides.setProperty(TopicConfig.RETENTION_BYTES_CONFIG, "100");
+        // Unknown topic config, but known broker config
+        topicOverrides.setProperty(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "sometrustpasswrd");
+        // Unknown config
+        topicOverrides.setProperty("unknown.topic.password.config", "bbbb");
+        // We don't currently have any sensitive topic configs, if we add them, we should set one here
+        LogConfig logConfig = LogConfig.fromProps(kafkaConfig.extractLogConfigMap(), topicOverrides);
+        assertEquals("{min.insync.replicas=2, retention.bytes=100, ssl.truststore.password=(redacted), unknown.topic.password.config=(redacted)}",
+                logConfig.overriddenConfigsAsLoggableString());
+    }
+
+    private boolean isValid(String configValue) {
+        try {
+            ThrottledReplicaListValidator.ensureValidString("", configValue);
+            return true;
+        } catch (ConfigException e) {
+            return false;
+        }
+    }
+
+    private KafkaConfig createKafkaConfig(boolean remoteStorageEnabled, Properties extra) {
+        Properties props = TestUtils.createDummyBrokerConfig();

Review Comment:
   also I can see that we have similar method in the jmh module (i.e., jmh-benchmarks/src/main/java/org/apache/kafka/jmh/util/BenchmarkConfigUtils.java) I wonder if those methods should be unified in some test-common sub-module  ... because it causing a lot of redudancy IMHO WDYT? 



-- 
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-20388: Move LogConfigTest from core to storage [kafka]

["m1a2st (via GitHub)" <gi...@apache.org>]

m1a2st commented on code in PR #21980:
URL: https://github.com/apache/kafka/pull/21980#discussion_r3073572927


##########
storage/src/test/java/org/apache/kafka/storage/internals/log/LogConfigTest.java:
##########
@@ -0,0 +1,460 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.storage.internals.log;
+
+import kafka.server.KafkaConfig;
+import kafka.utils.TestUtils;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.errors.InvalidConfigurationException;
+import org.apache.kafka.server.config.ServerLogConfigs;
+import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class LogConfigTest {
+
+    @Test
+    public void testKafkaConfigToProps() {
+        long millisInHour = 60L * 60L * 1000L;
+        long millisInDay = 24L * millisInHour;
+        long bytesInGB = 1024L * 1024L * 1024L;
+        Properties kafkaProps = TestUtils.createDummyBrokerConfig();
+        kafkaProps.put(ServerLogConfigs.LOG_ROLL_TIME_HOURS_CONFIG, "2");
+        kafkaProps.put(ServerLogConfigs.LOG_ROLL_TIME_JITTER_HOURS_CONFIG, "2");
+        kafkaProps.put(ServerLogConfigs.LOG_RETENTION_TIME_HOURS_CONFIG, "960"); // 40 days
+        kafkaProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, "2592000000"); // 30 days
+        kafkaProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, "4294967296"); // 4 GB
+
+        Map<String, Object> logProps = KafkaConfig.fromProps(kafkaProps).extractLogConfigMap();
+        assertEquals(2 * millisInHour, logProps.get(TopicConfig.SEGMENT_MS_CONFIG));
+        assertEquals(2 * millisInHour, logProps.get(TopicConfig.SEGMENT_JITTER_MS_CONFIG));
+        assertEquals(40 * millisInDay, logProps.get(TopicConfig.RETENTION_MS_CONFIG));
+        assertEquals(30 * millisInDay, logProps.get(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG));
+        assertEquals(4 * bytesInGB, logProps.get(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG));
+    }
+
+    @Test
+    public void testFromPropsInvalid() {
+        for (String name : LogConfig.configNames()) {
+            Object[] invalidValues = invalidValuesForProp(name);
+            if (invalidValues != null) {
+                assertPropertyInvalid(name, invalidValues);
+            }
+        }
+    }
+
+    private Object[] invalidValuesForProp(String name) {
+        return switch (name) {
+            case TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG -> new Object[]{"not a boolean"};
+            case TopicConfig.RETENTION_BYTES_CONFIG, TopicConfig.RETENTION_MS_CONFIG -> new Object[]{"not_a_number"};
+            case TopicConfig.CLEANUP_POLICY_CONFIG -> new Object[]{"true", "foobar"};
+            case TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG -> new Object[]{"not_a_number", "-0.1", "1.2"};
+            case TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG -> new Object[]{"not_a_number", "0", "-1"};
+            case TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG -> new Object[]{"not_a_number", "-3"};
+            case TopicConfig.COMPRESSION_GZIP_LEVEL_CONFIG -> new Object[]{"not_a_number", "-2"};
+            case TopicConfig.COMPRESSION_LZ4_LEVEL_CONFIG -> new Object[]{"not_a_number", "-1"};
+            case TopicConfig.COMPRESSION_ZSTD_LEVEL_CONFIG -> new Object[]{"not_a_number", "-0.1"};
+            case TopicConfig.REMOTE_LOG_COPY_DISABLE_CONFIG, TopicConfig.REMOTE_LOG_DELETE_ON_DISABLE_CONFIG -> new Object[]{"not_a_number", "remove", "0"};
+            case LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG -> null;
+            default -> new Object[]{"not_a_number", "-1"};
+        };
+    }
+
+    @Test
+    public void testInvalidCompactionLagConfig() {
+        HashMap<String, String> props = new HashMap<>();
+        props.put(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG, "100");
+        props.put(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG, "200");
+        assertThrows(Exception.class, () -> LogConfig.validate(props));
+    }
+
+    @Test
+    public void shouldValidateThrottledReplicasConfig() {
+        assertTrue(isValid("*"));
+        assertTrue(isValid("* "));
+        assertTrue(isValid(""));
+        assertTrue(isValid(" "));
+        assertTrue(isValid("100:10"));
+        assertTrue(isValid("100:10,12:10"));
+        assertTrue(isValid("100:10,12:10,15:1"));
+        assertTrue(isValid("100:10,12:10,15:1  "));
+        assertTrue(isValid("100:0,"));
+
+        assertFalse(isValid("100"));
+        assertFalse(isValid("100:"));
+        assertFalse(isValid("100:0,10"));
+        assertFalse(isValid("100:0,10:"));
+        assertFalse(isValid("100:0,10:   "));
+        assertFalse(isValid("100 :0,10:   "));
+        assertFalse(isValid("100: 0,10:   "));
+        assertFalse(isValid("100:0,10 :   "));
+        assertFalse(isValid("*,100:10"));
+        assertFalse(isValid("* ,100:10"));
+    }
+
+    /* Sanity check that toHtmlTable produces one of the expected configs */
+    @Test
+    public void testToHtmlTable() {
+        String html = LogConfig.configDefCopy().toHtmlTable();
+        String expectedConfig = "<td>file.delete.delay.ms</td>";
+        assertTrue(html.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + html);
+    }
+
+    /* Sanity check that toHtml produces one of the expected configs */
+    @Test
+    public void testToHtml() {
+        String html = LogConfig.configDefCopy().toHtml(4, key -> "prefix_" + key, Map.of());
+        String expectedConfig = "<h4><a id=\"file.delete.delay.ms\"></a><a id=\"prefix_file.delete.delay.ms\" href=\"#prefix_file.delete.delay.ms\">file.delete.delay.ms</a></h4>";
+        assertTrue(html.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + html);
+    }
+
+    /* Sanity check that toEnrichedRst produces one of the expected configs */
+    @Test
+    public void testToEnrichedRst() {
+        String rst = LogConfig.configDefCopy().toEnrichedRst();
+        String expectedConfig = "``file.delete.delay.ms``";
+        assertTrue(rst.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + rst);
+    }
+
+    /* Sanity check that toRst produces one of the expected configs */
+    @Test
+    public void testToRst() {
+        String rst = LogConfig.configDefCopy().toRst();
+        String expectedConfig = "``file.delete.delay.ms``";
+        assertTrue(rst.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + rst);
+    }
+
+    @Test
+    public void testGetConfigValue() {
+        // Add a config that doesn't set the `serverDefaultConfigName`
+        LogConfig.LogConfigDef configDef = LogConfig.configDefCopy();
+        String configNameWithNoServerMapping = "log.foo";
+        configDef.define(configNameWithNoServerMapping, ConfigDef.Type.INT, 1, ConfigDef.Importance.MEDIUM,
+                configNameWithNoServerMapping + " doc");
+
+        ConfigDef.ConfigKey deleteDelayKey = configDef.configKeys().get(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG);
+        Object deleteDelayServerDefault = configDef.getConfigValue(deleteDelayKey, LogConfig.SERVER_DEFAULT_HEADER_NAME);
+        assertEquals(ServerLogConfigs.LOG_DELETE_DELAY_MS_CONFIG, deleteDelayServerDefault);
+
+        ConfigDef.ConfigKey keyWithNoServerMapping = configDef.configKeys().get(configNameWithNoServerMapping);
+        Object nullServerDefault = configDef.getConfigValue(keyWithNoServerMapping, LogConfig.SERVER_DEFAULT_HEADER_NAME);
+        assertNull(nullServerDefault);
+    }
+
+    @Test
+    public void testOverriddenConfigsAsLoggableString() {
+        Properties kafkaProps = TestUtils.createDummyBrokerConfig();
+        kafkaProps.put("unknown.broker.password.config", "aaaaa");
+        kafkaProps.put(ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG, "50");
+        kafkaProps.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, "somekeypassword");
+        KafkaConfig kafkaConfig = KafkaConfig.fromProps(kafkaProps);
+        Properties topicOverrides = new Properties();
+        // Only set as a topic config
+        topicOverrides.setProperty(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2");
+        // Overrides value from broker config
+        topicOverrides.setProperty(TopicConfig.RETENTION_BYTES_CONFIG, "100");
+        // Unknown topic config, but known broker config
+        topicOverrides.setProperty(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "sometrustpasswrd");
+        // Unknown config
+        topicOverrides.setProperty("unknown.topic.password.config", "bbbb");
+        // We don't currently have any sensitive topic configs, if we add them, we should set one here
+        LogConfig logConfig = LogConfig.fromProps(kafkaConfig.extractLogConfigMap(), topicOverrides);
+        assertEquals("{min.insync.replicas=2, retention.bytes=100, ssl.truststore.password=(redacted), unknown.topic.password.config=(redacted)}",
+                logConfig.overriddenConfigsAsLoggableString());
+    }
+
+    private boolean isValid(String configValue) {
+        try {
+            ThrottledReplicaListValidator.ensureValidString("", configValue);
+            return true;
+        } catch (ConfigException e) {
+            return false;
+        }
+    }
+
+    private KafkaConfig createKafkaConfig(boolean remoteStorageEnabled, Properties extra) {
+        Properties props = TestUtils.createDummyBrokerConfig();

Review Comment:
   > I think I saw this inside RemoteLogManagerTest. But in the end then we should create some dummy helper method in the storage?
   
   +1
   
   > also I can see that we have similar method in the jmh module (i.e., jmh-benchmarks/src/main/java/org/apache/kafka/jmh/util/BenchmarkConfigUtils.java) I wonder if those methods should be unified in some test-common sub-module ... because it causing a lot of redudancy IMHO WDYT?
   
   Since this helper method isn’t widely applicable to all tests, I think it’s acceptable to keep some duplication across modules. The goal in [KAFKA-20350](https://issues.apache.org/jira/browse/KAFKA-20350) is to avoid test-common TestUtils becoming increasingly bloated.



-- 
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-20388: Move LogConfigTest from core to storage [kafka]

["see-quick (via GitHub)" <gi...@apache.org>]

see-quick commented on PR #21980:
URL: https://github.com/apache/kafka/pull/21980#issuecomment-4242142596

   > And if we still rely on Scala KafkaConfig, it may not be the right time to move this test.
   
   Regarding this ... there is `KafkaConfig.fromProps()` and specifically `extractLogConfigMap()` which would mean more movement but I assume it would be a bit more work ... Also there is clearYammerUtils but that should be handled by https://github.com/apache/kafka/pull/21679/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-20388: Move LogConfigTest from core to storage [kafka]

["m1a2st (via GitHub)" <gi...@apache.org>]

m1a2st commented on code in PR #21980:
URL: https://github.com/apache/kafka/pull/21980#discussion_r3051731715


##########
storage/src/test/java/org/apache/kafka/storage/internals/log/LogConfigTest.java:
##########
@@ -0,0 +1,460 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.storage.internals.log;
+
+import kafka.server.KafkaConfig;
+import kafka.utils.TestUtils;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.errors.InvalidConfigurationException;
+import org.apache.kafka.server.config.ServerLogConfigs;
+import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class LogConfigTest {
+
+    @Test
+    public void testKafkaConfigToProps() {
+        long millisInHour = 60L * 60L * 1000L;
+        long millisInDay = 24L * millisInHour;
+        long bytesInGB = 1024L * 1024L * 1024L;
+        Properties kafkaProps = TestUtils.createDummyBrokerConfig();
+        kafkaProps.put(ServerLogConfigs.LOG_ROLL_TIME_HOURS_CONFIG, "2");
+        kafkaProps.put(ServerLogConfigs.LOG_ROLL_TIME_JITTER_HOURS_CONFIG, "2");
+        kafkaProps.put(ServerLogConfigs.LOG_RETENTION_TIME_HOURS_CONFIG, "960"); // 40 days
+        kafkaProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, "2592000000"); // 30 days
+        kafkaProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, "4294967296"); // 4 GB
+
+        Map<String, Object> logProps = KafkaConfig.fromProps(kafkaProps).extractLogConfigMap();
+        assertEquals(2 * millisInHour, logProps.get(TopicConfig.SEGMENT_MS_CONFIG));
+        assertEquals(2 * millisInHour, logProps.get(TopicConfig.SEGMENT_JITTER_MS_CONFIG));
+        assertEquals(40 * millisInDay, logProps.get(TopicConfig.RETENTION_MS_CONFIG));
+        assertEquals(30 * millisInDay, logProps.get(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG));
+        assertEquals(4 * bytesInGB, logProps.get(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG));
+    }
+
+    @Test
+    public void testFromPropsInvalid() {
+        for (String name : LogConfig.configNames()) {
+            Object[] invalidValues = invalidValuesForProp(name);
+            if (invalidValues != null) {
+                assertPropertyInvalid(name, invalidValues);
+            }
+        }
+    }
+
+    private Object[] invalidValuesForProp(String name) {
+        return switch (name) {
+            case TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG -> new Object[]{"not a boolean"};
+            case TopicConfig.RETENTION_BYTES_CONFIG, TopicConfig.RETENTION_MS_CONFIG -> new Object[]{"not_a_number"};
+            case TopicConfig.CLEANUP_POLICY_CONFIG -> new Object[]{"true", "foobar"};
+            case TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG -> new Object[]{"not_a_number", "-0.1", "1.2"};
+            case TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG -> new Object[]{"not_a_number", "0", "-1"};
+            case TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG -> new Object[]{"not_a_number", "-3"};
+            case TopicConfig.COMPRESSION_GZIP_LEVEL_CONFIG -> new Object[]{"not_a_number", "-2"};
+            case TopicConfig.COMPRESSION_LZ4_LEVEL_CONFIG -> new Object[]{"not_a_number", "-1"};
+            case TopicConfig.COMPRESSION_ZSTD_LEVEL_CONFIG -> new Object[]{"not_a_number", "-0.1"};
+            case TopicConfig.REMOTE_LOG_COPY_DISABLE_CONFIG, TopicConfig.REMOTE_LOG_DELETE_ON_DISABLE_CONFIG -> new Object[]{"not_a_number", "remove", "0"};
+            case LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG -> null;
+            default -> new Object[]{"not_a_number", "-1"};
+        };
+    }
+
+    @Test
+    public void testInvalidCompactionLagConfig() {
+        HashMap<String, String> props = new HashMap<>();
+        props.put(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG, "100");
+        props.put(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG, "200");
+        assertThrows(Exception.class, () -> LogConfig.validate(props));
+    }
+
+    @Test
+    public void shouldValidateThrottledReplicasConfig() {
+        assertTrue(isValid("*"));
+        assertTrue(isValid("* "));
+        assertTrue(isValid(""));
+        assertTrue(isValid(" "));
+        assertTrue(isValid("100:10"));
+        assertTrue(isValid("100:10,12:10"));
+        assertTrue(isValid("100:10,12:10,15:1"));
+        assertTrue(isValid("100:10,12:10,15:1  "));
+        assertTrue(isValid("100:0,"));
+
+        assertFalse(isValid("100"));
+        assertFalse(isValid("100:"));
+        assertFalse(isValid("100:0,10"));
+        assertFalse(isValid("100:0,10:"));
+        assertFalse(isValid("100:0,10:   "));
+        assertFalse(isValid("100 :0,10:   "));
+        assertFalse(isValid("100: 0,10:   "));
+        assertFalse(isValid("100:0,10 :   "));
+        assertFalse(isValid("*,100:10"));
+        assertFalse(isValid("* ,100:10"));
+    }
+
+    /* Sanity check that toHtmlTable produces one of the expected configs */
+    @Test
+    public void testToHtmlTable() {
+        String html = LogConfig.configDefCopy().toHtmlTable();
+        String expectedConfig = "<td>file.delete.delay.ms</td>";
+        assertTrue(html.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + html);
+    }
+
+    /* Sanity check that toHtml produces one of the expected configs */
+    @Test
+    public void testToHtml() {
+        String html = LogConfig.configDefCopy().toHtml(4, key -> "prefix_" + key, Map.of());
+        String expectedConfig = "<h4><a id=\"file.delete.delay.ms\"></a><a id=\"prefix_file.delete.delay.ms\" href=\"#prefix_file.delete.delay.ms\">file.delete.delay.ms</a></h4>";
+        assertTrue(html.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + html);
+    }
+
+    /* Sanity check that toEnrichedRst produces one of the expected configs */
+    @Test
+    public void testToEnrichedRst() {
+        String rst = LogConfig.configDefCopy().toEnrichedRst();
+        String expectedConfig = "``file.delete.delay.ms``";
+        assertTrue(rst.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + rst);
+    }
+
+    /* Sanity check that toRst produces one of the expected configs */
+    @Test
+    public void testToRst() {
+        String rst = LogConfig.configDefCopy().toRst();
+        String expectedConfig = "``file.delete.delay.ms``";
+        assertTrue(rst.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + rst);
+    }
+
+    @Test
+    public void testGetConfigValue() {
+        // Add a config that doesn't set the `serverDefaultConfigName`
+        LogConfig.LogConfigDef configDef = LogConfig.configDefCopy();
+        String configNameWithNoServerMapping = "log.foo";
+        configDef.define(configNameWithNoServerMapping, ConfigDef.Type.INT, 1, ConfigDef.Importance.MEDIUM,
+                configNameWithNoServerMapping + " doc");
+
+        ConfigDef.ConfigKey deleteDelayKey = configDef.configKeys().get(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG);
+        Object deleteDelayServerDefault = configDef.getConfigValue(deleteDelayKey, LogConfig.SERVER_DEFAULT_HEADER_NAME);
+        assertEquals(ServerLogConfigs.LOG_DELETE_DELAY_MS_CONFIG, deleteDelayServerDefault);
+
+        ConfigDef.ConfigKey keyWithNoServerMapping = configDef.configKeys().get(configNameWithNoServerMapping);
+        Object nullServerDefault = configDef.getConfigValue(keyWithNoServerMapping, LogConfig.SERVER_DEFAULT_HEADER_NAME);
+        assertNull(nullServerDefault);
+    }
+
+    @Test
+    public void testOverriddenConfigsAsLoggableString() {
+        Properties kafkaProps = TestUtils.createDummyBrokerConfig();
+        kafkaProps.put("unknown.broker.password.config", "aaaaa");
+        kafkaProps.put(ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG, "50");
+        kafkaProps.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, "somekeypassword");
+        KafkaConfig kafkaConfig = KafkaConfig.fromProps(kafkaProps);
+        Properties topicOverrides = new Properties();
+        // Only set as a topic config
+        topicOverrides.setProperty(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2");
+        // Overrides value from broker config
+        topicOverrides.setProperty(TopicConfig.RETENTION_BYTES_CONFIG, "100");
+        // Unknown topic config, but known broker config
+        topicOverrides.setProperty(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "sometrustpasswrd");
+        // Unknown config
+        topicOverrides.setProperty("unknown.topic.password.config", "bbbb");
+        // We don't currently have any sensitive topic configs, if we add them, we should set one here
+        LogConfig logConfig = LogConfig.fromProps(kafkaConfig.extractLogConfigMap(), topicOverrides);
+        assertEquals("{min.insync.replicas=2, retention.bytes=100, ssl.truststore.password=(redacted), unknown.topic.password.config=(redacted)}",
+                logConfig.overriddenConfigsAsLoggableString());
+    }
+
+    private boolean isValid(String configValue) {
+        try {
+            ThrottledReplicaListValidator.ensureValidString("", configValue);
+            return true;
+        } catch (ConfigException e) {
+            return false;
+        }
+    }
+
+    private KafkaConfig createKafkaConfig(boolean remoteStorageEnabled, Properties extra) {
+        Properties props = TestUtils.createDummyBrokerConfig();
+        props.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, String.valueOf(remoteStorageEnabled));
+        props.putAll(extra);
+        return KafkaConfig.fromProps(props);
+    }
+
+    private KafkaConfig createKafkaConfig(boolean remoteStorageEnabled) {
+        return createKafkaConfig(remoteStorageEnabled, new Properties());
+    }
+
+    private void assertPropertyInvalid(String name, Object... values) {
+        for (Object value : values) {
+            Properties props = new Properties();
+            props.setProperty(name, value.toString());
+            assertThrows(Exception.class, () -> new LogConfig(props),
+                () -> "Property " + name + " should not allow " + value);
+        }
+    }
+
+    @Test
+    public void testLocalLogRetentionDerivedProps() {
+        Properties props = new Properties();
+        int retentionBytes = 1024;
+        long retentionMs = 1000L;
+        props.put(TopicConfig.RETENTION_BYTES_CONFIG, String.valueOf(retentionBytes));
+        props.put(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(retentionMs));
+        LogConfig logConfig = new LogConfig(props);
+
+        assertEquals(retentionMs, logConfig.localRetentionMs());
+        assertEquals(retentionBytes, logConfig.localRetentionBytes());
+    }
+
+    @Test
+    public void testLocalLogRetentionDerivedDefaultProps() {
+        LogConfig logConfig = new LogConfig(new Properties());
+
+        // Local retention defaults are derived from retention properties which can be default or custom.
+        assertEquals(LogConfig.DEFAULT_RETENTION_MS, logConfig.localRetentionMs());
+        assertEquals(ServerLogConfigs.LOG_RETENTION_BYTES_DEFAULT, logConfig.localRetentionBytes());
+    }
+
+    @Test
+    public void testLocalLogRetentionProps() {
+        Properties props = new Properties();
+        int localRetentionMs = 500;
+        int localRetentionBytes = 1000;
+        props.put(TopicConfig.RETENTION_BYTES_CONFIG, "2000");
+        props.put(TopicConfig.RETENTION_MS_CONFIG, "1000");
+
+        props.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, String.valueOf(localRetentionMs));
+        props.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, String.valueOf(localRetentionBytes));
+        LogConfig logConfig = new LogConfig(props);
+
+        assertEquals(localRetentionMs, logConfig.localRetentionMs());
+        assertEquals(localRetentionBytes, logConfig.localRetentionBytes());
+    }
+
+    @Test
+    public void testInvalidLocalLogRetentionProps() {
+        // Check for invalid localRetentionMs, < -2
+        doTestInvalidLocalLogRetentionProps(-3, 10, 2, 500L);
+
+        // Check for invalid localRetentionBytes < -2
+        doTestInvalidLocalLogRetentionProps(500L, -3, 2, 1000L);
+
+        // Check for invalid case of localRetentionMs > retentionMs
+        doTestInvalidLocalLogRetentionProps(2000L, 2, 100, 1000L);
+
+        // Check for invalid case of localRetentionBytes > retentionBytes
+        doTestInvalidLocalLogRetentionProps(500L, 200, 100, 1000L);
+
+        // Check for invalid case of localRetentionMs (-1 viz unlimited) > retentionMs,
+        doTestInvalidLocalLogRetentionProps(-1, 200, 100, 1000L);
+
+        // Check for invalid case of localRetentionBytes(-1 viz unlimited) > retentionBytes
+        doTestInvalidLocalLogRetentionProps(2000L, -1, 100, 1000L);
+    }
+
+    private void doTestInvalidLocalLogRetentionProps(long localRetentionMs,
+                                                     int localRetentionBytes,
+                                                     int retentionBytes,
+                                                     long retentionMs) {
+        KafkaConfig kafkaConfig = createKafkaConfig(true);
+
+        HashMap<String, String> props = new HashMap<>();
+        props.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true");
+        props.put(TopicConfig.RETENTION_BYTES_CONFIG, String.valueOf(retentionBytes));
+        props.put(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(retentionMs));
+
+        props.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, String.valueOf(localRetentionMs));
+        props.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, String.valueOf(localRetentionBytes));
+        assertThrows(ConfigException.class,
+            () -> LogConfig.validate(Map.of(), props, kafkaConfig.extractLogConfigMap(),
+                kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+    }
+
+    @Test
+    public void testEnableRemoteLogStorageCleanupPolicy() {
+        KafkaConfig kafkaConfig = createKafkaConfig(true);
+        HashMap<String, String> logProps = new HashMap<>();
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE);
+        logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true");
+        LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT);
+        assertThrows(ConfigException.class, () -> LogConfig.validate(Map.of(), logProps,
+                kafkaConfig.extractLogConfigMap(), kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "delete,compact");
+        assertThrows(ConfigException.class, () -> LogConfig.validate(Map.of(), logProps,
+                kafkaConfig.extractLogConfigMap(), kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "compact,delete");
+        assertThrows(ConfigException.class, () -> LogConfig.validate(Map.of(), logProps,
+                kafkaConfig.extractLogConfigMap(), kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "delete,delete,delete");
+        LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "");
+        LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+    }
+
+    @ParameterizedTest(name = "testEnableRemoteLogStorage with sysRemoteStorageEnabled: {0}")
+    @ValueSource(booleans = {true, false})
+    public void testEnableRemoteLogStorage(boolean sysRemoteStorageEnabled) {
+        KafkaConfig kafkaConfig = createKafkaConfig(sysRemoteStorageEnabled);
+
+        HashMap<String, String> logProps = new HashMap<>();
+        logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true");

Review Comment:
   ```suggestion
           Map<String, String> logProps = Map.of(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true");
   ```



##########
storage/src/test/java/org/apache/kafka/storage/internals/log/LogConfigTest.java:
##########
@@ -0,0 +1,460 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.storage.internals.log;
+
+import kafka.server.KafkaConfig;
+import kafka.utils.TestUtils;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.errors.InvalidConfigurationException;
+import org.apache.kafka.server.config.ServerLogConfigs;
+import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class LogConfigTest {
+
+    @Test
+    public void testKafkaConfigToProps() {
+        long millisInHour = 60L * 60L * 1000L;
+        long millisInDay = 24L * millisInHour;
+        long bytesInGB = 1024L * 1024L * 1024L;
+        Properties kafkaProps = TestUtils.createDummyBrokerConfig();
+        kafkaProps.put(ServerLogConfigs.LOG_ROLL_TIME_HOURS_CONFIG, "2");
+        kafkaProps.put(ServerLogConfigs.LOG_ROLL_TIME_JITTER_HOURS_CONFIG, "2");
+        kafkaProps.put(ServerLogConfigs.LOG_RETENTION_TIME_HOURS_CONFIG, "960"); // 40 days
+        kafkaProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, "2592000000"); // 30 days
+        kafkaProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, "4294967296"); // 4 GB
+
+        Map<String, Object> logProps = KafkaConfig.fromProps(kafkaProps).extractLogConfigMap();
+        assertEquals(2 * millisInHour, logProps.get(TopicConfig.SEGMENT_MS_CONFIG));
+        assertEquals(2 * millisInHour, logProps.get(TopicConfig.SEGMENT_JITTER_MS_CONFIG));
+        assertEquals(40 * millisInDay, logProps.get(TopicConfig.RETENTION_MS_CONFIG));
+        assertEquals(30 * millisInDay, logProps.get(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG));
+        assertEquals(4 * bytesInGB, logProps.get(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG));
+    }
+
+    @Test
+    public void testFromPropsInvalid() {
+        for (String name : LogConfig.configNames()) {
+            Object[] invalidValues = invalidValuesForProp(name);
+            if (invalidValues != null) {
+                assertPropertyInvalid(name, invalidValues);
+            }
+        }
+    }
+
+    private Object[] invalidValuesForProp(String name) {
+        return switch (name) {
+            case TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG -> new Object[]{"not a boolean"};
+            case TopicConfig.RETENTION_BYTES_CONFIG, TopicConfig.RETENTION_MS_CONFIG -> new Object[]{"not_a_number"};
+            case TopicConfig.CLEANUP_POLICY_CONFIG -> new Object[]{"true", "foobar"};
+            case TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG -> new Object[]{"not_a_number", "-0.1", "1.2"};
+            case TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG -> new Object[]{"not_a_number", "0", "-1"};
+            case TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG -> new Object[]{"not_a_number", "-3"};
+            case TopicConfig.COMPRESSION_GZIP_LEVEL_CONFIG -> new Object[]{"not_a_number", "-2"};
+            case TopicConfig.COMPRESSION_LZ4_LEVEL_CONFIG -> new Object[]{"not_a_number", "-1"};
+            case TopicConfig.COMPRESSION_ZSTD_LEVEL_CONFIG -> new Object[]{"not_a_number", "-0.1"};
+            case TopicConfig.REMOTE_LOG_COPY_DISABLE_CONFIG, TopicConfig.REMOTE_LOG_DELETE_ON_DISABLE_CONFIG -> new Object[]{"not_a_number", "remove", "0"};
+            case LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG -> null;
+            default -> new Object[]{"not_a_number", "-1"};
+        };
+    }
+
+    @Test
+    public void testInvalidCompactionLagConfig() {
+        HashMap<String, String> props = new HashMap<>();
+        props.put(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG, "100");
+        props.put(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG, "200");
+        assertThrows(Exception.class, () -> LogConfig.validate(props));
+    }
+
+    @Test
+    public void shouldValidateThrottledReplicasConfig() {
+        assertTrue(isValid("*"));
+        assertTrue(isValid("* "));
+        assertTrue(isValid(""));
+        assertTrue(isValid(" "));
+        assertTrue(isValid("100:10"));
+        assertTrue(isValid("100:10,12:10"));
+        assertTrue(isValid("100:10,12:10,15:1"));
+        assertTrue(isValid("100:10,12:10,15:1  "));
+        assertTrue(isValid("100:0,"));
+
+        assertFalse(isValid("100"));
+        assertFalse(isValid("100:"));
+        assertFalse(isValid("100:0,10"));
+        assertFalse(isValid("100:0,10:"));
+        assertFalse(isValid("100:0,10:   "));
+        assertFalse(isValid("100 :0,10:   "));
+        assertFalse(isValid("100: 0,10:   "));
+        assertFalse(isValid("100:0,10 :   "));
+        assertFalse(isValid("*,100:10"));
+        assertFalse(isValid("* ,100:10"));
+    }
+
+    /* Sanity check that toHtmlTable produces one of the expected configs */
+    @Test
+    public void testToHtmlTable() {
+        String html = LogConfig.configDefCopy().toHtmlTable();
+        String expectedConfig = "<td>file.delete.delay.ms</td>";
+        assertTrue(html.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + html);
+    }
+
+    /* Sanity check that toHtml produces one of the expected configs */
+    @Test
+    public void testToHtml() {
+        String html = LogConfig.configDefCopy().toHtml(4, key -> "prefix_" + key, Map.of());
+        String expectedConfig = "<h4><a id=\"file.delete.delay.ms\"></a><a id=\"prefix_file.delete.delay.ms\" href=\"#prefix_file.delete.delay.ms\">file.delete.delay.ms</a></h4>";
+        assertTrue(html.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + html);
+    }
+
+    /* Sanity check that toEnrichedRst produces one of the expected configs */
+    @Test
+    public void testToEnrichedRst() {
+        String rst = LogConfig.configDefCopy().toEnrichedRst();
+        String expectedConfig = "``file.delete.delay.ms``";
+        assertTrue(rst.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + rst);
+    }
+
+    /* Sanity check that toRst produces one of the expected configs */
+    @Test
+    public void testToRst() {
+        String rst = LogConfig.configDefCopy().toRst();
+        String expectedConfig = "``file.delete.delay.ms``";
+        assertTrue(rst.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + rst);
+    }
+
+    @Test
+    public void testGetConfigValue() {
+        // Add a config that doesn't set the `serverDefaultConfigName`
+        LogConfig.LogConfigDef configDef = LogConfig.configDefCopy();
+        String configNameWithNoServerMapping = "log.foo";
+        configDef.define(configNameWithNoServerMapping, ConfigDef.Type.INT, 1, ConfigDef.Importance.MEDIUM,
+                configNameWithNoServerMapping + " doc");
+
+        ConfigDef.ConfigKey deleteDelayKey = configDef.configKeys().get(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG);
+        Object deleteDelayServerDefault = configDef.getConfigValue(deleteDelayKey, LogConfig.SERVER_DEFAULT_HEADER_NAME);
+        assertEquals(ServerLogConfigs.LOG_DELETE_DELAY_MS_CONFIG, deleteDelayServerDefault);
+
+        ConfigDef.ConfigKey keyWithNoServerMapping = configDef.configKeys().get(configNameWithNoServerMapping);
+        Object nullServerDefault = configDef.getConfigValue(keyWithNoServerMapping, LogConfig.SERVER_DEFAULT_HEADER_NAME);
+        assertNull(nullServerDefault);
+    }
+
+    @Test
+    public void testOverriddenConfigsAsLoggableString() {
+        Properties kafkaProps = TestUtils.createDummyBrokerConfig();
+        kafkaProps.put("unknown.broker.password.config", "aaaaa");
+        kafkaProps.put(ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG, "50");
+        kafkaProps.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, "somekeypassword");
+        KafkaConfig kafkaConfig = KafkaConfig.fromProps(kafkaProps);
+        Properties topicOverrides = new Properties();
+        // Only set as a topic config
+        topicOverrides.setProperty(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2");
+        // Overrides value from broker config
+        topicOverrides.setProperty(TopicConfig.RETENTION_BYTES_CONFIG, "100");
+        // Unknown topic config, but known broker config
+        topicOverrides.setProperty(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "sometrustpasswrd");
+        // Unknown config
+        topicOverrides.setProperty("unknown.topic.password.config", "bbbb");
+        // We don't currently have any sensitive topic configs, if we add them, we should set one here
+        LogConfig logConfig = LogConfig.fromProps(kafkaConfig.extractLogConfigMap(), topicOverrides);
+        assertEquals("{min.insync.replicas=2, retention.bytes=100, ssl.truststore.password=(redacted), unknown.topic.password.config=(redacted)}",
+                logConfig.overriddenConfigsAsLoggableString());
+    }
+
+    private boolean isValid(String configValue) {
+        try {
+            ThrottledReplicaListValidator.ensureValidString("", configValue);
+            return true;
+        } catch (ConfigException e) {
+            return false;
+        }
+    }
+
+    private KafkaConfig createKafkaConfig(boolean remoteStorageEnabled, Properties extra) {
+        Properties props = TestUtils.createDummyBrokerConfig();
+        props.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, String.valueOf(remoteStorageEnabled));
+        props.putAll(extra);
+        return KafkaConfig.fromProps(props);
+    }
+
+    private KafkaConfig createKafkaConfig(boolean remoteStorageEnabled) {
+        return createKafkaConfig(remoteStorageEnabled, new Properties());
+    }
+
+    private void assertPropertyInvalid(String name, Object... values) {
+        for (Object value : values) {
+            Properties props = new Properties();
+            props.setProperty(name, value.toString());
+            assertThrows(Exception.class, () -> new LogConfig(props),
+                () -> "Property " + name + " should not allow " + value);
+        }
+    }
+
+    @Test
+    public void testLocalLogRetentionDerivedProps() {
+        Properties props = new Properties();
+        int retentionBytes = 1024;
+        long retentionMs = 1000L;
+        props.put(TopicConfig.RETENTION_BYTES_CONFIG, String.valueOf(retentionBytes));
+        props.put(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(retentionMs));
+        LogConfig logConfig = new LogConfig(props);
+
+        assertEquals(retentionMs, logConfig.localRetentionMs());
+        assertEquals(retentionBytes, logConfig.localRetentionBytes());
+    }
+
+    @Test
+    public void testLocalLogRetentionDerivedDefaultProps() {
+        LogConfig logConfig = new LogConfig(new Properties());
+
+        // Local retention defaults are derived from retention properties which can be default or custom.
+        assertEquals(LogConfig.DEFAULT_RETENTION_MS, logConfig.localRetentionMs());
+        assertEquals(ServerLogConfigs.LOG_RETENTION_BYTES_DEFAULT, logConfig.localRetentionBytes());
+    }
+
+    @Test
+    public void testLocalLogRetentionProps() {
+        Properties props = new Properties();
+        int localRetentionMs = 500;
+        int localRetentionBytes = 1000;
+        props.put(TopicConfig.RETENTION_BYTES_CONFIG, "2000");
+        props.put(TopicConfig.RETENTION_MS_CONFIG, "1000");
+
+        props.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, String.valueOf(localRetentionMs));
+        props.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, String.valueOf(localRetentionBytes));
+        LogConfig logConfig = new LogConfig(props);
+
+        assertEquals(localRetentionMs, logConfig.localRetentionMs());
+        assertEquals(localRetentionBytes, logConfig.localRetentionBytes());
+    }
+
+    @Test
+    public void testInvalidLocalLogRetentionProps() {
+        // Check for invalid localRetentionMs, < -2
+        doTestInvalidLocalLogRetentionProps(-3, 10, 2, 500L);
+
+        // Check for invalid localRetentionBytes < -2
+        doTestInvalidLocalLogRetentionProps(500L, -3, 2, 1000L);
+
+        // Check for invalid case of localRetentionMs > retentionMs
+        doTestInvalidLocalLogRetentionProps(2000L, 2, 100, 1000L);
+
+        // Check for invalid case of localRetentionBytes > retentionBytes
+        doTestInvalidLocalLogRetentionProps(500L, 200, 100, 1000L);
+
+        // Check for invalid case of localRetentionMs (-1 viz unlimited) > retentionMs,
+        doTestInvalidLocalLogRetentionProps(-1, 200, 100, 1000L);
+
+        // Check for invalid case of localRetentionBytes(-1 viz unlimited) > retentionBytes
+        doTestInvalidLocalLogRetentionProps(2000L, -1, 100, 1000L);
+    }
+
+    private void doTestInvalidLocalLogRetentionProps(long localRetentionMs,
+                                                     int localRetentionBytes,
+                                                     int retentionBytes,
+                                                     long retentionMs) {
+        KafkaConfig kafkaConfig = createKafkaConfig(true);
+
+        HashMap<String, String> props = new HashMap<>();
+        props.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true");
+        props.put(TopicConfig.RETENTION_BYTES_CONFIG, String.valueOf(retentionBytes));
+        props.put(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(retentionMs));
+
+        props.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, String.valueOf(localRetentionMs));
+        props.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, String.valueOf(localRetentionBytes));
+        assertThrows(ConfigException.class,
+            () -> LogConfig.validate(Map.of(), props, kafkaConfig.extractLogConfigMap(),
+                kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+    }
+
+    @Test
+    public void testEnableRemoteLogStorageCleanupPolicy() {
+        KafkaConfig kafkaConfig = createKafkaConfig(true);
+        HashMap<String, String> logProps = new HashMap<>();
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE);
+        logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true");
+        LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT);
+        assertThrows(ConfigException.class, () -> LogConfig.validate(Map.of(), logProps,
+                kafkaConfig.extractLogConfigMap(), kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "delete,compact");
+        assertThrows(ConfigException.class, () -> LogConfig.validate(Map.of(), logProps,
+                kafkaConfig.extractLogConfigMap(), kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "compact,delete");
+        assertThrows(ConfigException.class, () -> LogConfig.validate(Map.of(), logProps,
+                kafkaConfig.extractLogConfigMap(), kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "delete,delete,delete");
+        LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "");
+        LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+    }
+
+    @ParameterizedTest(name = "testEnableRemoteLogStorage with sysRemoteStorageEnabled: {0}")
+    @ValueSource(booleans = {true, false})
+    public void testEnableRemoteLogStorage(boolean sysRemoteStorageEnabled) {
+        KafkaConfig kafkaConfig = createKafkaConfig(sysRemoteStorageEnabled);
+
+        HashMap<String, String> logProps = new HashMap<>();
+        logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true");
+        if (sysRemoteStorageEnabled) {
+            LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                    kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+        } else {
+            ConfigException message = assertThrows(ConfigException.class,
+                    () -> LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                            kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+            assertTrue(message.getMessage().contains("Tiered Storage functionality is disabled in the broker"));
+        }
+    }
+
+    @ParameterizedTest(name = "testDisableRemoteLogStorage with wasRemoteStorageEnabled: {0}")
+    @ValueSource(booleans = {true, false})
+    public void testDisableRemoteLogStorage(boolean wasRemoteStorageEnabled) {
+        KafkaConfig kafkaConfig = createKafkaConfig(true);
+
+        HashMap<String, String> logProps = new HashMap<>();
+        logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "false");
+        if (wasRemoteStorageEnabled) {
+            InvalidConfigurationException message = assertThrows(InvalidConfigurationException.class,
+                    () -> LogConfig.validate(Map.of(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true"),
+                            logProps, kafkaConfig.extractLogConfigMap(),
+                            kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+            assertTrue(message.getMessage().contains("It is invalid to disable remote storage without deleting remote data. " +
+                    "If you want to keep the remote data and turn to read only, please set `remote.storage.enable=true,remote.log.copy.disable=true`. " +
+                    "If you want to disable remote storage and delete all remote data, please set `remote.storage.enable=false,remote.log.delete.on.disable=true`."));
+
+            // It should be able to disable the remote log storage when delete on disable is set to true
+            logProps.put(TopicConfig.REMOTE_LOG_DELETE_ON_DISABLE_CONFIG, "true");
+            LogConfig.validate(Map.of(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true"),
+                    logProps, kafkaConfig.extractLogConfigMap(),
+                    kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+        } else {
+            LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                    kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+            LogConfig.validate(Map.of(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "false"), logProps,
+                    kafkaConfig.extractLogConfigMap(), kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+        }
+    }
+
+    @ParameterizedTest(name = "testTopicCreationWithInvalidRetentionTime with sysRemoteStorageEnabled: {0}")
+    @ValueSource(booleans = {true, false})
+    public void testTopicCreationWithInvalidRetentionTime(boolean sysRemoteStorageEnabled) {
+        Properties extra = new Properties();
+        extra.put(ServerLogConfigs.LOG_RETENTION_TIME_MILLIS_CONFIG, "1000");
+        extra.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, "900");
+        KafkaConfig kafkaConfig = createKafkaConfig(sysRemoteStorageEnabled, extra);
+
+        // Topic local log retention time inherited from Broker is greater than the topic's complete log retention time
+        HashMap<String, String> logProps = new HashMap<>();
+        logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, String.valueOf(sysRemoteStorageEnabled));
+        logProps.put(TopicConfig.RETENTION_MS_CONFIG, "500");
+        if (sysRemoteStorageEnabled) {
+            ConfigException message = assertThrows(ConfigException.class,
+                    () -> LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                            kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+            assertTrue(message.getMessage().contains(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG));
+        } else {
+            LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                    kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+        }
+    }
+
+    @ParameterizedTest(name = "testTopicCreationWithInvalidRetentionSize with sysRemoteStorageEnabled: {0}")
+    @ValueSource(booleans = {true, false})
+    public void testTopicCreationWithInvalidRetentionSize(boolean sysRemoteStorageEnabled) {
+        Properties extra = new Properties();
+        extra.put(ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG, "1024");
+        extra.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, "512");
+        KafkaConfig kafkaConfig = createKafkaConfig(sysRemoteStorageEnabled, extra);
+
+        // Topic local retention size inherited from Broker is greater than the topic's complete log retention size
+        HashMap<String, String> logProps = new HashMap<>();
+        logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, String.valueOf(sysRemoteStorageEnabled));
+        logProps.put(TopicConfig.RETENTION_BYTES_CONFIG, "128");
+        if (sysRemoteStorageEnabled) {
+            ConfigException message = assertThrows(ConfigException.class,
+                    () -> LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                            kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+            assertTrue(message.getMessage().contains(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG));
+        } else {
+            LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                    kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+        }
+    }
+
+    @ParameterizedTest(name = "testValidateBrokerLogConfigs with sysRemoteStorageEnabled: {0}")
+    @ValueSource(booleans = {true, false})
+    public void testValidateBrokerLogConfigs(boolean sysRemoteStorageEnabled) {
+        Properties extra = new Properties();
+        extra.put(ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG, "1024");
+        extra.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, "2048");
+        KafkaConfig kafkaConfig = createKafkaConfig(sysRemoteStorageEnabled, extra);
+
+        if (sysRemoteStorageEnabled) {
+            ConfigException message = assertThrows(ConfigException.class,
+                    () -> LogConfig.validateBrokerLogConfigValues(kafkaConfig.extractLogConfigMap(),
+                            kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+            assertTrue(message.getMessage().contains(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG));
+        } else {
+            LogConfig.validateBrokerLogConfigValues(kafkaConfig.extractLogConfigMap(),
+                    kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+        }
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void testValidRemoteLogCopyDisabled(boolean copyDisabled) {
+        HashMap<String, String> logProps = new HashMap<>();
+        logProps.put(TopicConfig.REMOTE_LOG_COPY_DISABLE_CONFIG, String.valueOf(copyDisabled));

Review Comment:
   ```suggestion
           Map<String, String> logProps = Map.of(TopicConfig.REMOTE_LOG_COPY_DISABLE_CONFIG, String.valueOf(copyDisabled));
   ```



##########
storage/src/test/java/org/apache/kafka/storage/internals/log/LogConfigTest.java:
##########
@@ -0,0 +1,460 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.storage.internals.log;
+
+import kafka.server.KafkaConfig;
+import kafka.utils.TestUtils;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.errors.InvalidConfigurationException;
+import org.apache.kafka.server.config.ServerLogConfigs;
+import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class LogConfigTest {
+
+    @Test
+    public void testKafkaConfigToProps() {
+        long millisInHour = 60L * 60L * 1000L;
+        long millisInDay = 24L * millisInHour;
+        long bytesInGB = 1024L * 1024L * 1024L;
+        Properties kafkaProps = TestUtils.createDummyBrokerConfig();
+        kafkaProps.put(ServerLogConfigs.LOG_ROLL_TIME_HOURS_CONFIG, "2");
+        kafkaProps.put(ServerLogConfigs.LOG_ROLL_TIME_JITTER_HOURS_CONFIG, "2");
+        kafkaProps.put(ServerLogConfigs.LOG_RETENTION_TIME_HOURS_CONFIG, "960"); // 40 days
+        kafkaProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, "2592000000"); // 30 days
+        kafkaProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, "4294967296"); // 4 GB
+
+        Map<String, Object> logProps = KafkaConfig.fromProps(kafkaProps).extractLogConfigMap();
+        assertEquals(2 * millisInHour, logProps.get(TopicConfig.SEGMENT_MS_CONFIG));
+        assertEquals(2 * millisInHour, logProps.get(TopicConfig.SEGMENT_JITTER_MS_CONFIG));
+        assertEquals(40 * millisInDay, logProps.get(TopicConfig.RETENTION_MS_CONFIG));
+        assertEquals(30 * millisInDay, logProps.get(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG));
+        assertEquals(4 * bytesInGB, logProps.get(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG));
+    }
+
+    @Test
+    public void testFromPropsInvalid() {
+        for (String name : LogConfig.configNames()) {
+            Object[] invalidValues = invalidValuesForProp(name);
+            if (invalidValues != null) {
+                assertPropertyInvalid(name, invalidValues);
+            }
+        }
+    }
+
+    private Object[] invalidValuesForProp(String name) {
+        return switch (name) {
+            case TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG -> new Object[]{"not a boolean"};
+            case TopicConfig.RETENTION_BYTES_CONFIG, TopicConfig.RETENTION_MS_CONFIG -> new Object[]{"not_a_number"};
+            case TopicConfig.CLEANUP_POLICY_CONFIG -> new Object[]{"true", "foobar"};
+            case TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG -> new Object[]{"not_a_number", "-0.1", "1.2"};
+            case TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG -> new Object[]{"not_a_number", "0", "-1"};
+            case TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG -> new Object[]{"not_a_number", "-3"};
+            case TopicConfig.COMPRESSION_GZIP_LEVEL_CONFIG -> new Object[]{"not_a_number", "-2"};
+            case TopicConfig.COMPRESSION_LZ4_LEVEL_CONFIG -> new Object[]{"not_a_number", "-1"};
+            case TopicConfig.COMPRESSION_ZSTD_LEVEL_CONFIG -> new Object[]{"not_a_number", "-0.1"};
+            case TopicConfig.REMOTE_LOG_COPY_DISABLE_CONFIG, TopicConfig.REMOTE_LOG_DELETE_ON_DISABLE_CONFIG -> new Object[]{"not_a_number", "remove", "0"};
+            case LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG -> null;
+            default -> new Object[]{"not_a_number", "-1"};
+        };
+    }
+
+    @Test
+    public void testInvalidCompactionLagConfig() {
+        HashMap<String, String> props = new HashMap<>();
+        props.put(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG, "100");
+        props.put(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG, "200");
+        assertThrows(Exception.class, () -> LogConfig.validate(props));
+    }
+
+    @Test
+    public void shouldValidateThrottledReplicasConfig() {
+        assertTrue(isValid("*"));
+        assertTrue(isValid("* "));
+        assertTrue(isValid(""));
+        assertTrue(isValid(" "));
+        assertTrue(isValid("100:10"));
+        assertTrue(isValid("100:10,12:10"));
+        assertTrue(isValid("100:10,12:10,15:1"));
+        assertTrue(isValid("100:10,12:10,15:1  "));
+        assertTrue(isValid("100:0,"));
+
+        assertFalse(isValid("100"));
+        assertFalse(isValid("100:"));
+        assertFalse(isValid("100:0,10"));
+        assertFalse(isValid("100:0,10:"));
+        assertFalse(isValid("100:0,10:   "));
+        assertFalse(isValid("100 :0,10:   "));
+        assertFalse(isValid("100: 0,10:   "));
+        assertFalse(isValid("100:0,10 :   "));
+        assertFalse(isValid("*,100:10"));
+        assertFalse(isValid("* ,100:10"));
+    }
+
+    /* Sanity check that toHtmlTable produces one of the expected configs */
+    @Test
+    public void testToHtmlTable() {
+        String html = LogConfig.configDefCopy().toHtmlTable();
+        String expectedConfig = "<td>file.delete.delay.ms</td>";
+        assertTrue(html.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + html);
+    }
+
+    /* Sanity check that toHtml produces one of the expected configs */
+    @Test
+    public void testToHtml() {
+        String html = LogConfig.configDefCopy().toHtml(4, key -> "prefix_" + key, Map.of());
+        String expectedConfig = "<h4><a id=\"file.delete.delay.ms\"></a><a id=\"prefix_file.delete.delay.ms\" href=\"#prefix_file.delete.delay.ms\">file.delete.delay.ms</a></h4>";
+        assertTrue(html.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + html);
+    }
+
+    /* Sanity check that toEnrichedRst produces one of the expected configs */
+    @Test
+    public void testToEnrichedRst() {
+        String rst = LogConfig.configDefCopy().toEnrichedRst();
+        String expectedConfig = "``file.delete.delay.ms``";
+        assertTrue(rst.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + rst);
+    }
+
+    /* Sanity check that toRst produces one of the expected configs */
+    @Test
+    public void testToRst() {
+        String rst = LogConfig.configDefCopy().toRst();
+        String expectedConfig = "``file.delete.delay.ms``";
+        assertTrue(rst.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + rst);
+    }
+
+    @Test
+    public void testGetConfigValue() {
+        // Add a config that doesn't set the `serverDefaultConfigName`
+        LogConfig.LogConfigDef configDef = LogConfig.configDefCopy();
+        String configNameWithNoServerMapping = "log.foo";
+        configDef.define(configNameWithNoServerMapping, ConfigDef.Type.INT, 1, ConfigDef.Importance.MEDIUM,
+                configNameWithNoServerMapping + " doc");
+
+        ConfigDef.ConfigKey deleteDelayKey = configDef.configKeys().get(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG);
+        Object deleteDelayServerDefault = configDef.getConfigValue(deleteDelayKey, LogConfig.SERVER_DEFAULT_HEADER_NAME);
+        assertEquals(ServerLogConfigs.LOG_DELETE_DELAY_MS_CONFIG, deleteDelayServerDefault);
+
+        ConfigDef.ConfigKey keyWithNoServerMapping = configDef.configKeys().get(configNameWithNoServerMapping);
+        Object nullServerDefault = configDef.getConfigValue(keyWithNoServerMapping, LogConfig.SERVER_DEFAULT_HEADER_NAME);
+        assertNull(nullServerDefault);
+    }
+
+    @Test
+    public void testOverriddenConfigsAsLoggableString() {
+        Properties kafkaProps = TestUtils.createDummyBrokerConfig();
+        kafkaProps.put("unknown.broker.password.config", "aaaaa");
+        kafkaProps.put(ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG, "50");
+        kafkaProps.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, "somekeypassword");
+        KafkaConfig kafkaConfig = KafkaConfig.fromProps(kafkaProps);
+        Properties topicOverrides = new Properties();
+        // Only set as a topic config
+        topicOverrides.setProperty(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2");
+        // Overrides value from broker config
+        topicOverrides.setProperty(TopicConfig.RETENTION_BYTES_CONFIG, "100");
+        // Unknown topic config, but known broker config
+        topicOverrides.setProperty(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "sometrustpasswrd");
+        // Unknown config
+        topicOverrides.setProperty("unknown.topic.password.config", "bbbb");
+        // We don't currently have any sensitive topic configs, if we add them, we should set one here
+        LogConfig logConfig = LogConfig.fromProps(kafkaConfig.extractLogConfigMap(), topicOverrides);
+        assertEquals("{min.insync.replicas=2, retention.bytes=100, ssl.truststore.password=(redacted), unknown.topic.password.config=(redacted)}",
+                logConfig.overriddenConfigsAsLoggableString());
+    }
+
+    private boolean isValid(String configValue) {
+        try {
+            ThrottledReplicaListValidator.ensureValidString("", configValue);
+            return true;
+        } catch (ConfigException e) {
+            return false;
+        }
+    }
+
+    private KafkaConfig createKafkaConfig(boolean remoteStorageEnabled, Properties extra) {
+        Properties props = TestUtils.createDummyBrokerConfig();

Review Comment:
   Java tests should avoid to depend on Scala `TestUtils`.



##########
storage/src/test/java/org/apache/kafka/storage/internals/log/LogConfigTest.java:
##########
@@ -0,0 +1,460 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.storage.internals.log;
+
+import kafka.server.KafkaConfig;
+import kafka.utils.TestUtils;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.errors.InvalidConfigurationException;
+import org.apache.kafka.server.config.ServerLogConfigs;
+import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class LogConfigTest {
+
+    @Test
+    public void testKafkaConfigToProps() {
+        long millisInHour = 60L * 60L * 1000L;
+        long millisInDay = 24L * millisInHour;
+        long bytesInGB = 1024L * 1024L * 1024L;
+        Properties kafkaProps = TestUtils.createDummyBrokerConfig();
+        kafkaProps.put(ServerLogConfigs.LOG_ROLL_TIME_HOURS_CONFIG, "2");
+        kafkaProps.put(ServerLogConfigs.LOG_ROLL_TIME_JITTER_HOURS_CONFIG, "2");
+        kafkaProps.put(ServerLogConfigs.LOG_RETENTION_TIME_HOURS_CONFIG, "960"); // 40 days
+        kafkaProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, "2592000000"); // 30 days
+        kafkaProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, "4294967296"); // 4 GB
+
+        Map<String, Object> logProps = KafkaConfig.fromProps(kafkaProps).extractLogConfigMap();
+        assertEquals(2 * millisInHour, logProps.get(TopicConfig.SEGMENT_MS_CONFIG));
+        assertEquals(2 * millisInHour, logProps.get(TopicConfig.SEGMENT_JITTER_MS_CONFIG));
+        assertEquals(40 * millisInDay, logProps.get(TopicConfig.RETENTION_MS_CONFIG));
+        assertEquals(30 * millisInDay, logProps.get(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG));
+        assertEquals(4 * bytesInGB, logProps.get(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG));
+    }
+
+    @Test
+    public void testFromPropsInvalid() {
+        for (String name : LogConfig.configNames()) {
+            Object[] invalidValues = invalidValuesForProp(name);
+            if (invalidValues != null) {
+                assertPropertyInvalid(name, invalidValues);
+            }
+        }
+    }
+
+    private Object[] invalidValuesForProp(String name) {
+        return switch (name) {
+            case TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG -> new Object[]{"not a boolean"};
+            case TopicConfig.RETENTION_BYTES_CONFIG, TopicConfig.RETENTION_MS_CONFIG -> new Object[]{"not_a_number"};
+            case TopicConfig.CLEANUP_POLICY_CONFIG -> new Object[]{"true", "foobar"};
+            case TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG -> new Object[]{"not_a_number", "-0.1", "1.2"};
+            case TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG -> new Object[]{"not_a_number", "0", "-1"};
+            case TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG -> new Object[]{"not_a_number", "-3"};
+            case TopicConfig.COMPRESSION_GZIP_LEVEL_CONFIG -> new Object[]{"not_a_number", "-2"};
+            case TopicConfig.COMPRESSION_LZ4_LEVEL_CONFIG -> new Object[]{"not_a_number", "-1"};
+            case TopicConfig.COMPRESSION_ZSTD_LEVEL_CONFIG -> new Object[]{"not_a_number", "-0.1"};
+            case TopicConfig.REMOTE_LOG_COPY_DISABLE_CONFIG, TopicConfig.REMOTE_LOG_DELETE_ON_DISABLE_CONFIG -> new Object[]{"not_a_number", "remove", "0"};
+            case LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG -> null;
+            default -> new Object[]{"not_a_number", "-1"};
+        };
+    }
+
+    @Test
+    public void testInvalidCompactionLagConfig() {
+        HashMap<String, String> props = new HashMap<>();
+        props.put(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG, "100");
+        props.put(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG, "200");
+        assertThrows(Exception.class, () -> LogConfig.validate(props));
+    }
+
+    @Test
+    public void shouldValidateThrottledReplicasConfig() {
+        assertTrue(isValid("*"));
+        assertTrue(isValid("* "));
+        assertTrue(isValid(""));
+        assertTrue(isValid(" "));
+        assertTrue(isValid("100:10"));
+        assertTrue(isValid("100:10,12:10"));
+        assertTrue(isValid("100:10,12:10,15:1"));
+        assertTrue(isValid("100:10,12:10,15:1  "));
+        assertTrue(isValid("100:0,"));
+
+        assertFalse(isValid("100"));
+        assertFalse(isValid("100:"));
+        assertFalse(isValid("100:0,10"));
+        assertFalse(isValid("100:0,10:"));
+        assertFalse(isValid("100:0,10:   "));
+        assertFalse(isValid("100 :0,10:   "));
+        assertFalse(isValid("100: 0,10:   "));
+        assertFalse(isValid("100:0,10 :   "));
+        assertFalse(isValid("*,100:10"));
+        assertFalse(isValid("* ,100:10"));
+    }
+
+    /* Sanity check that toHtmlTable produces one of the expected configs */
+    @Test
+    public void testToHtmlTable() {
+        String html = LogConfig.configDefCopy().toHtmlTable();
+        String expectedConfig = "<td>file.delete.delay.ms</td>";
+        assertTrue(html.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + html);
+    }
+
+    /* Sanity check that toHtml produces one of the expected configs */
+    @Test
+    public void testToHtml() {
+        String html = LogConfig.configDefCopy().toHtml(4, key -> "prefix_" + key, Map.of());
+        String expectedConfig = "<h4><a id=\"file.delete.delay.ms\"></a><a id=\"prefix_file.delete.delay.ms\" href=\"#prefix_file.delete.delay.ms\">file.delete.delay.ms</a></h4>";
+        assertTrue(html.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + html);
+    }
+
+    /* Sanity check that toEnrichedRst produces one of the expected configs */
+    @Test
+    public void testToEnrichedRst() {
+        String rst = LogConfig.configDefCopy().toEnrichedRst();
+        String expectedConfig = "``file.delete.delay.ms``";
+        assertTrue(rst.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + rst);
+    }
+
+    /* Sanity check that toRst produces one of the expected configs */
+    @Test
+    public void testToRst() {
+        String rst = LogConfig.configDefCopy().toRst();
+        String expectedConfig = "``file.delete.delay.ms``";
+        assertTrue(rst.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + rst);
+    }
+
+    @Test
+    public void testGetConfigValue() {
+        // Add a config that doesn't set the `serverDefaultConfigName`
+        LogConfig.LogConfigDef configDef = LogConfig.configDefCopy();
+        String configNameWithNoServerMapping = "log.foo";
+        configDef.define(configNameWithNoServerMapping, ConfigDef.Type.INT, 1, ConfigDef.Importance.MEDIUM,
+                configNameWithNoServerMapping + " doc");
+
+        ConfigDef.ConfigKey deleteDelayKey = configDef.configKeys().get(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG);
+        Object deleteDelayServerDefault = configDef.getConfigValue(deleteDelayKey, LogConfig.SERVER_DEFAULT_HEADER_NAME);
+        assertEquals(ServerLogConfigs.LOG_DELETE_DELAY_MS_CONFIG, deleteDelayServerDefault);
+
+        ConfigDef.ConfigKey keyWithNoServerMapping = configDef.configKeys().get(configNameWithNoServerMapping);
+        Object nullServerDefault = configDef.getConfigValue(keyWithNoServerMapping, LogConfig.SERVER_DEFAULT_HEADER_NAME);
+        assertNull(nullServerDefault);
+    }
+
+    @Test
+    public void testOverriddenConfigsAsLoggableString() {
+        Properties kafkaProps = TestUtils.createDummyBrokerConfig();
+        kafkaProps.put("unknown.broker.password.config", "aaaaa");
+        kafkaProps.put(ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG, "50");
+        kafkaProps.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, "somekeypassword");
+        KafkaConfig kafkaConfig = KafkaConfig.fromProps(kafkaProps);
+        Properties topicOverrides = new Properties();
+        // Only set as a topic config
+        topicOverrides.setProperty(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2");
+        // Overrides value from broker config
+        topicOverrides.setProperty(TopicConfig.RETENTION_BYTES_CONFIG, "100");
+        // Unknown topic config, but known broker config
+        topicOverrides.setProperty(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "sometrustpasswrd");
+        // Unknown config
+        topicOverrides.setProperty("unknown.topic.password.config", "bbbb");
+        // We don't currently have any sensitive topic configs, if we add them, we should set one here
+        LogConfig logConfig = LogConfig.fromProps(kafkaConfig.extractLogConfigMap(), topicOverrides);
+        assertEquals("{min.insync.replicas=2, retention.bytes=100, ssl.truststore.password=(redacted), unknown.topic.password.config=(redacted)}",
+                logConfig.overriddenConfigsAsLoggableString());
+    }
+
+    private boolean isValid(String configValue) {
+        try {
+            ThrottledReplicaListValidator.ensureValidString("", configValue);
+            return true;
+        } catch (ConfigException e) {
+            return false;
+        }
+    }
+
+    private KafkaConfig createKafkaConfig(boolean remoteStorageEnabled, Properties extra) {
+        Properties props = TestUtils.createDummyBrokerConfig();
+        props.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, String.valueOf(remoteStorageEnabled));
+        props.putAll(extra);
+        return KafkaConfig.fromProps(props);
+    }
+
+    private KafkaConfig createKafkaConfig(boolean remoteStorageEnabled) {
+        return createKafkaConfig(remoteStorageEnabled, new Properties());
+    }
+
+    private void assertPropertyInvalid(String name, Object... values) {
+        for (Object value : values) {
+            Properties props = new Properties();
+            props.setProperty(name, value.toString());
+            assertThrows(Exception.class, () -> new LogConfig(props),
+                () -> "Property " + name + " should not allow " + value);
+        }
+    }
+
+    @Test
+    public void testLocalLogRetentionDerivedProps() {
+        Properties props = new Properties();
+        int retentionBytes = 1024;
+        long retentionMs = 1000L;
+        props.put(TopicConfig.RETENTION_BYTES_CONFIG, String.valueOf(retentionBytes));
+        props.put(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(retentionMs));
+        LogConfig logConfig = new LogConfig(props);
+
+        assertEquals(retentionMs, logConfig.localRetentionMs());
+        assertEquals(retentionBytes, logConfig.localRetentionBytes());
+    }
+
+    @Test
+    public void testLocalLogRetentionDerivedDefaultProps() {
+        LogConfig logConfig = new LogConfig(new Properties());
+
+        // Local retention defaults are derived from retention properties which can be default or custom.
+        assertEquals(LogConfig.DEFAULT_RETENTION_MS, logConfig.localRetentionMs());
+        assertEquals(ServerLogConfigs.LOG_RETENTION_BYTES_DEFAULT, logConfig.localRetentionBytes());
+    }
+
+    @Test
+    public void testLocalLogRetentionProps() {
+        Properties props = new Properties();
+        int localRetentionMs = 500;
+        int localRetentionBytes = 1000;
+        props.put(TopicConfig.RETENTION_BYTES_CONFIG, "2000");
+        props.put(TopicConfig.RETENTION_MS_CONFIG, "1000");
+
+        props.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, String.valueOf(localRetentionMs));
+        props.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, String.valueOf(localRetentionBytes));
+        LogConfig logConfig = new LogConfig(props);
+
+        assertEquals(localRetentionMs, logConfig.localRetentionMs());
+        assertEquals(localRetentionBytes, logConfig.localRetentionBytes());
+    }
+
+    @Test
+    public void testInvalidLocalLogRetentionProps() {
+        // Check for invalid localRetentionMs, < -2
+        doTestInvalidLocalLogRetentionProps(-3, 10, 2, 500L);
+
+        // Check for invalid localRetentionBytes < -2
+        doTestInvalidLocalLogRetentionProps(500L, -3, 2, 1000L);
+
+        // Check for invalid case of localRetentionMs > retentionMs
+        doTestInvalidLocalLogRetentionProps(2000L, 2, 100, 1000L);
+
+        // Check for invalid case of localRetentionBytes > retentionBytes
+        doTestInvalidLocalLogRetentionProps(500L, 200, 100, 1000L);
+
+        // Check for invalid case of localRetentionMs (-1 viz unlimited) > retentionMs,
+        doTestInvalidLocalLogRetentionProps(-1, 200, 100, 1000L);
+
+        // Check for invalid case of localRetentionBytes(-1 viz unlimited) > retentionBytes
+        doTestInvalidLocalLogRetentionProps(2000L, -1, 100, 1000L);
+    }
+
+    private void doTestInvalidLocalLogRetentionProps(long localRetentionMs,
+                                                     int localRetentionBytes,
+                                                     int retentionBytes,
+                                                     long retentionMs) {
+        KafkaConfig kafkaConfig = createKafkaConfig(true);
+
+        HashMap<String, String> props = new HashMap<>();
+        props.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true");
+        props.put(TopicConfig.RETENTION_BYTES_CONFIG, String.valueOf(retentionBytes));
+        props.put(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(retentionMs));
+
+        props.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, String.valueOf(localRetentionMs));
+        props.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, String.valueOf(localRetentionBytes));
+        assertThrows(ConfigException.class,
+            () -> LogConfig.validate(Map.of(), props, kafkaConfig.extractLogConfigMap(),
+                kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+    }
+
+    @Test
+    public void testEnableRemoteLogStorageCleanupPolicy() {
+        KafkaConfig kafkaConfig = createKafkaConfig(true);
+        HashMap<String, String> logProps = new HashMap<>();
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE);
+        logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true");
+        LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT);
+        assertThrows(ConfigException.class, () -> LogConfig.validate(Map.of(), logProps,
+                kafkaConfig.extractLogConfigMap(), kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "delete,compact");
+        assertThrows(ConfigException.class, () -> LogConfig.validate(Map.of(), logProps,
+                kafkaConfig.extractLogConfigMap(), kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "compact,delete");
+        assertThrows(ConfigException.class, () -> LogConfig.validate(Map.of(), logProps,
+                kafkaConfig.extractLogConfigMap(), kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "delete,delete,delete");
+        LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "");
+        LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+    }
+
+    @ParameterizedTest(name = "testEnableRemoteLogStorage with sysRemoteStorageEnabled: {0}")
+    @ValueSource(booleans = {true, false})
+    public void testEnableRemoteLogStorage(boolean sysRemoteStorageEnabled) {
+        KafkaConfig kafkaConfig = createKafkaConfig(sysRemoteStorageEnabled);
+
+        HashMap<String, String> logProps = new HashMap<>();
+        logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true");
+        if (sysRemoteStorageEnabled) {
+            LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                    kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+        } else {
+            ConfigException message = assertThrows(ConfigException.class,
+                    () -> LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                            kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+            assertTrue(message.getMessage().contains("Tiered Storage functionality is disabled in the broker"));
+        }
+    }
+
+    @ParameterizedTest(name = "testDisableRemoteLogStorage with wasRemoteStorageEnabled: {0}")
+    @ValueSource(booleans = {true, false})
+    public void testDisableRemoteLogStorage(boolean wasRemoteStorageEnabled) {
+        KafkaConfig kafkaConfig = createKafkaConfig(true);
+
+        HashMap<String, String> logProps = new HashMap<>();
+        logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "false");
+        if (wasRemoteStorageEnabled) {
+            InvalidConfigurationException message = assertThrows(InvalidConfigurationException.class,
+                    () -> LogConfig.validate(Map.of(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true"),
+                            logProps, kafkaConfig.extractLogConfigMap(),
+                            kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+            assertTrue(message.getMessage().contains("It is invalid to disable remote storage without deleting remote data. " +
+                    "If you want to keep the remote data and turn to read only, please set `remote.storage.enable=true,remote.log.copy.disable=true`. " +
+                    "If you want to disable remote storage and delete all remote data, please set `remote.storage.enable=false,remote.log.delete.on.disable=true`."));
+
+            // It should be able to disable the remote log storage when delete on disable is set to true
+            logProps.put(TopicConfig.REMOTE_LOG_DELETE_ON_DISABLE_CONFIG, "true");
+            LogConfig.validate(Map.of(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true"),
+                    logProps, kafkaConfig.extractLogConfigMap(),
+                    kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+        } else {
+            LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                    kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+            LogConfig.validate(Map.of(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "false"), logProps,
+                    kafkaConfig.extractLogConfigMap(), kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+        }
+    }
+
+    @ParameterizedTest(name = "testTopicCreationWithInvalidRetentionTime with sysRemoteStorageEnabled: {0}")
+    @ValueSource(booleans = {true, false})
+    public void testTopicCreationWithInvalidRetentionTime(boolean sysRemoteStorageEnabled) {
+        Properties extra = new Properties();
+        extra.put(ServerLogConfigs.LOG_RETENTION_TIME_MILLIS_CONFIG, "1000");
+        extra.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, "900");
+        KafkaConfig kafkaConfig = createKafkaConfig(sysRemoteStorageEnabled, extra);
+
+        // Topic local log retention time inherited from Broker is greater than the topic's complete log retention time
+        HashMap<String, String> logProps = new HashMap<>();
+        logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, String.valueOf(sysRemoteStorageEnabled));
+        logProps.put(TopicConfig.RETENTION_MS_CONFIG, "500");
+        if (sysRemoteStorageEnabled) {
+            ConfigException message = assertThrows(ConfigException.class,
+                    () -> LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                            kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+            assertTrue(message.getMessage().contains(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG));
+        } else {
+            LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                    kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+        }
+    }
+
+    @ParameterizedTest(name = "testTopicCreationWithInvalidRetentionSize with sysRemoteStorageEnabled: {0}")
+    @ValueSource(booleans = {true, false})
+    public void testTopicCreationWithInvalidRetentionSize(boolean sysRemoteStorageEnabled) {
+        Properties extra = new Properties();
+        extra.put(ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG, "1024");
+        extra.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, "512");
+        KafkaConfig kafkaConfig = createKafkaConfig(sysRemoteStorageEnabled, extra);
+
+        // Topic local retention size inherited from Broker is greater than the topic's complete log retention size
+        HashMap<String, String> logProps = new HashMap<>();
+        logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, String.valueOf(sysRemoteStorageEnabled));
+        logProps.put(TopicConfig.RETENTION_BYTES_CONFIG, "128");
+        if (sysRemoteStorageEnabled) {
+            ConfigException message = assertThrows(ConfigException.class,
+                    () -> LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                            kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+            assertTrue(message.getMessage().contains(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG));
+        } else {
+            LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                    kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+        }
+    }
+
+    @ParameterizedTest(name = "testValidateBrokerLogConfigs with sysRemoteStorageEnabled: {0}")
+    @ValueSource(booleans = {true, false})
+    public void testValidateBrokerLogConfigs(boolean sysRemoteStorageEnabled) {
+        Properties extra = new Properties();
+        extra.put(ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG, "1024");
+        extra.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, "2048");
+        KafkaConfig kafkaConfig = createKafkaConfig(sysRemoteStorageEnabled, extra);
+
+        if (sysRemoteStorageEnabled) {
+            ConfigException message = assertThrows(ConfigException.class,
+                    () -> LogConfig.validateBrokerLogConfigValues(kafkaConfig.extractLogConfigMap(),
+                            kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+            assertTrue(message.getMessage().contains(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG));
+        } else {
+            LogConfig.validateBrokerLogConfigValues(kafkaConfig.extractLogConfigMap(),
+                    kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+        }
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void testValidRemoteLogCopyDisabled(boolean copyDisabled) {
+        HashMap<String, String> logProps = new HashMap<>();
+        logProps.put(TopicConfig.REMOTE_LOG_COPY_DISABLE_CONFIG, String.valueOf(copyDisabled));
+        LogConfig.validate(logProps);
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void testValidRemoteLogDeleteOnDisable(boolean deleteOnDisable) {
+        HashMap<String, String> logProps = new HashMap<>();
+        logProps.put(TopicConfig.REMOTE_LOG_DELETE_ON_DISABLE_CONFIG, String.valueOf(deleteOnDisable));

Review Comment:
   ```suggestion
           Map<String, String> logProps = Map.of(TopicConfig.REMOTE_LOG_DELETE_ON_DISABLE_CONFIG, String.valueOf(deleteOnDisable));
   ```



##########
storage/src/test/java/org/apache/kafka/storage/internals/log/LogConfigTest.java:
##########
@@ -0,0 +1,460 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.storage.internals.log;
+
+import kafka.server.KafkaConfig;
+import kafka.utils.TestUtils;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.errors.InvalidConfigurationException;
+import org.apache.kafka.server.config.ServerLogConfigs;
+import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class LogConfigTest {
+
+    @Test
+    public void testKafkaConfigToProps() {
+        long millisInHour = 60L * 60L * 1000L;
+        long millisInDay = 24L * millisInHour;
+        long bytesInGB = 1024L * 1024L * 1024L;
+        Properties kafkaProps = TestUtils.createDummyBrokerConfig();
+        kafkaProps.put(ServerLogConfigs.LOG_ROLL_TIME_HOURS_CONFIG, "2");
+        kafkaProps.put(ServerLogConfigs.LOG_ROLL_TIME_JITTER_HOURS_CONFIG, "2");
+        kafkaProps.put(ServerLogConfigs.LOG_RETENTION_TIME_HOURS_CONFIG, "960"); // 40 days
+        kafkaProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, "2592000000"); // 30 days
+        kafkaProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, "4294967296"); // 4 GB
+
+        Map<String, Object> logProps = KafkaConfig.fromProps(kafkaProps).extractLogConfigMap();
+        assertEquals(2 * millisInHour, logProps.get(TopicConfig.SEGMENT_MS_CONFIG));
+        assertEquals(2 * millisInHour, logProps.get(TopicConfig.SEGMENT_JITTER_MS_CONFIG));
+        assertEquals(40 * millisInDay, logProps.get(TopicConfig.RETENTION_MS_CONFIG));
+        assertEquals(30 * millisInDay, logProps.get(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG));
+        assertEquals(4 * bytesInGB, logProps.get(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG));
+    }
+
+    @Test
+    public void testFromPropsInvalid() {
+        for (String name : LogConfig.configNames()) {
+            Object[] invalidValues = invalidValuesForProp(name);
+            if (invalidValues != null) {
+                assertPropertyInvalid(name, invalidValues);
+            }
+        }
+    }
+
+    private Object[] invalidValuesForProp(String name) {
+        return switch (name) {
+            case TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG -> new Object[]{"not a boolean"};
+            case TopicConfig.RETENTION_BYTES_CONFIG, TopicConfig.RETENTION_MS_CONFIG -> new Object[]{"not_a_number"};
+            case TopicConfig.CLEANUP_POLICY_CONFIG -> new Object[]{"true", "foobar"};
+            case TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG -> new Object[]{"not_a_number", "-0.1", "1.2"};
+            case TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG -> new Object[]{"not_a_number", "0", "-1"};
+            case TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG -> new Object[]{"not_a_number", "-3"};
+            case TopicConfig.COMPRESSION_GZIP_LEVEL_CONFIG -> new Object[]{"not_a_number", "-2"};
+            case TopicConfig.COMPRESSION_LZ4_LEVEL_CONFIG -> new Object[]{"not_a_number", "-1"};
+            case TopicConfig.COMPRESSION_ZSTD_LEVEL_CONFIG -> new Object[]{"not_a_number", "-0.1"};
+            case TopicConfig.REMOTE_LOG_COPY_DISABLE_CONFIG, TopicConfig.REMOTE_LOG_DELETE_ON_DISABLE_CONFIG -> new Object[]{"not_a_number", "remove", "0"};
+            case LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG -> null;
+            default -> new Object[]{"not_a_number", "-1"};
+        };
+    }
+
+    @Test
+    public void testInvalidCompactionLagConfig() {
+        HashMap<String, String> props = new HashMap<>();
+        props.put(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG, "100");
+        props.put(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG, "200");
+        assertThrows(Exception.class, () -> LogConfig.validate(props));
+    }
+
+    @Test
+    public void shouldValidateThrottledReplicasConfig() {
+        assertTrue(isValid("*"));
+        assertTrue(isValid("* "));
+        assertTrue(isValid(""));
+        assertTrue(isValid(" "));
+        assertTrue(isValid("100:10"));
+        assertTrue(isValid("100:10,12:10"));
+        assertTrue(isValid("100:10,12:10,15:1"));
+        assertTrue(isValid("100:10,12:10,15:1  "));
+        assertTrue(isValid("100:0,"));
+
+        assertFalse(isValid("100"));
+        assertFalse(isValid("100:"));
+        assertFalse(isValid("100:0,10"));
+        assertFalse(isValid("100:0,10:"));
+        assertFalse(isValid("100:0,10:   "));
+        assertFalse(isValid("100 :0,10:   "));
+        assertFalse(isValid("100: 0,10:   "));
+        assertFalse(isValid("100:0,10 :   "));
+        assertFalse(isValid("*,100:10"));
+        assertFalse(isValid("* ,100:10"));
+    }
+
+    /* Sanity check that toHtmlTable produces one of the expected configs */
+    @Test
+    public void testToHtmlTable() {
+        String html = LogConfig.configDefCopy().toHtmlTable();
+        String expectedConfig = "<td>file.delete.delay.ms</td>";
+        assertTrue(html.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + html);
+    }
+
+    /* Sanity check that toHtml produces one of the expected configs */
+    @Test
+    public void testToHtml() {
+        String html = LogConfig.configDefCopy().toHtml(4, key -> "prefix_" + key, Map.of());
+        String expectedConfig = "<h4><a id=\"file.delete.delay.ms\"></a><a id=\"prefix_file.delete.delay.ms\" href=\"#prefix_file.delete.delay.ms\">file.delete.delay.ms</a></h4>";
+        assertTrue(html.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + html);
+    }
+
+    /* Sanity check that toEnrichedRst produces one of the expected configs */
+    @Test
+    public void testToEnrichedRst() {
+        String rst = LogConfig.configDefCopy().toEnrichedRst();
+        String expectedConfig = "``file.delete.delay.ms``";
+        assertTrue(rst.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + rst);
+    }
+
+    /* Sanity check that toRst produces one of the expected configs */
+    @Test
+    public void testToRst() {
+        String rst = LogConfig.configDefCopy().toRst();
+        String expectedConfig = "``file.delete.delay.ms``";
+        assertTrue(rst.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + rst);
+    }
+
+    @Test
+    public void testGetConfigValue() {
+        // Add a config that doesn't set the `serverDefaultConfigName`
+        LogConfig.LogConfigDef configDef = LogConfig.configDefCopy();
+        String configNameWithNoServerMapping = "log.foo";
+        configDef.define(configNameWithNoServerMapping, ConfigDef.Type.INT, 1, ConfigDef.Importance.MEDIUM,
+                configNameWithNoServerMapping + " doc");
+
+        ConfigDef.ConfigKey deleteDelayKey = configDef.configKeys().get(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG);
+        Object deleteDelayServerDefault = configDef.getConfigValue(deleteDelayKey, LogConfig.SERVER_DEFAULT_HEADER_NAME);
+        assertEquals(ServerLogConfigs.LOG_DELETE_DELAY_MS_CONFIG, deleteDelayServerDefault);
+
+        ConfigDef.ConfigKey keyWithNoServerMapping = configDef.configKeys().get(configNameWithNoServerMapping);
+        Object nullServerDefault = configDef.getConfigValue(keyWithNoServerMapping, LogConfig.SERVER_DEFAULT_HEADER_NAME);
+        assertNull(nullServerDefault);
+    }
+
+    @Test
+    public void testOverriddenConfigsAsLoggableString() {
+        Properties kafkaProps = TestUtils.createDummyBrokerConfig();
+        kafkaProps.put("unknown.broker.password.config", "aaaaa");
+        kafkaProps.put(ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG, "50");
+        kafkaProps.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, "somekeypassword");
+        KafkaConfig kafkaConfig = KafkaConfig.fromProps(kafkaProps);
+        Properties topicOverrides = new Properties();
+        // Only set as a topic config
+        topicOverrides.setProperty(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2");
+        // Overrides value from broker config
+        topicOverrides.setProperty(TopicConfig.RETENTION_BYTES_CONFIG, "100");
+        // Unknown topic config, but known broker config
+        topicOverrides.setProperty(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "sometrustpasswrd");
+        // Unknown config
+        topicOverrides.setProperty("unknown.topic.password.config", "bbbb");
+        // We don't currently have any sensitive topic configs, if we add them, we should set one here
+        LogConfig logConfig = LogConfig.fromProps(kafkaConfig.extractLogConfigMap(), topicOverrides);
+        assertEquals("{min.insync.replicas=2, retention.bytes=100, ssl.truststore.password=(redacted), unknown.topic.password.config=(redacted)}",
+                logConfig.overriddenConfigsAsLoggableString());
+    }
+
+    private boolean isValid(String configValue) {
+        try {
+            ThrottledReplicaListValidator.ensureValidString("", configValue);
+            return true;
+        } catch (ConfigException e) {
+            return false;
+        }
+    }
+
+    private KafkaConfig createKafkaConfig(boolean remoteStorageEnabled, Properties extra) {
+        Properties props = TestUtils.createDummyBrokerConfig();
+        props.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, String.valueOf(remoteStorageEnabled));
+        props.putAll(extra);
+        return KafkaConfig.fromProps(props);
+    }
+
+    private KafkaConfig createKafkaConfig(boolean remoteStorageEnabled) {
+        return createKafkaConfig(remoteStorageEnabled, new Properties());
+    }
+
+    private void assertPropertyInvalid(String name, Object... values) {
+        for (Object value : values) {
+            Properties props = new Properties();
+            props.setProperty(name, value.toString());
+            assertThrows(Exception.class, () -> new LogConfig(props),
+                () -> "Property " + name + " should not allow " + value);
+        }
+    }
+
+    @Test
+    public void testLocalLogRetentionDerivedProps() {
+        Properties props = new Properties();
+        int retentionBytes = 1024;
+        long retentionMs = 1000L;
+        props.put(TopicConfig.RETENTION_BYTES_CONFIG, String.valueOf(retentionBytes));
+        props.put(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(retentionMs));
+        LogConfig logConfig = new LogConfig(props);
+
+        assertEquals(retentionMs, logConfig.localRetentionMs());
+        assertEquals(retentionBytes, logConfig.localRetentionBytes());
+    }
+
+    @Test
+    public void testLocalLogRetentionDerivedDefaultProps() {
+        LogConfig logConfig = new LogConfig(new Properties());
+
+        // Local retention defaults are derived from retention properties which can be default or custom.
+        assertEquals(LogConfig.DEFAULT_RETENTION_MS, logConfig.localRetentionMs());
+        assertEquals(ServerLogConfigs.LOG_RETENTION_BYTES_DEFAULT, logConfig.localRetentionBytes());
+    }
+
+    @Test
+    public void testLocalLogRetentionProps() {
+        Properties props = new Properties();
+        int localRetentionMs = 500;
+        int localRetentionBytes = 1000;
+        props.put(TopicConfig.RETENTION_BYTES_CONFIG, "2000");
+        props.put(TopicConfig.RETENTION_MS_CONFIG, "1000");
+
+        props.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, String.valueOf(localRetentionMs));
+        props.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, String.valueOf(localRetentionBytes));
+        LogConfig logConfig = new LogConfig(props);
+
+        assertEquals(localRetentionMs, logConfig.localRetentionMs());
+        assertEquals(localRetentionBytes, logConfig.localRetentionBytes());
+    }
+
+    @Test
+    public void testInvalidLocalLogRetentionProps() {
+        // Check for invalid localRetentionMs, < -2
+        doTestInvalidLocalLogRetentionProps(-3, 10, 2, 500L);
+
+        // Check for invalid localRetentionBytes < -2
+        doTestInvalidLocalLogRetentionProps(500L, -3, 2, 1000L);
+
+        // Check for invalid case of localRetentionMs > retentionMs
+        doTestInvalidLocalLogRetentionProps(2000L, 2, 100, 1000L);
+
+        // Check for invalid case of localRetentionBytes > retentionBytes
+        doTestInvalidLocalLogRetentionProps(500L, 200, 100, 1000L);
+
+        // Check for invalid case of localRetentionMs (-1 viz unlimited) > retentionMs,
+        doTestInvalidLocalLogRetentionProps(-1, 200, 100, 1000L);
+
+        // Check for invalid case of localRetentionBytes(-1 viz unlimited) > retentionBytes
+        doTestInvalidLocalLogRetentionProps(2000L, -1, 100, 1000L);
+    }
+
+    private void doTestInvalidLocalLogRetentionProps(long localRetentionMs,
+                                                     int localRetentionBytes,
+                                                     int retentionBytes,
+                                                     long retentionMs) {
+        KafkaConfig kafkaConfig = createKafkaConfig(true);
+
+        HashMap<String, String> props = new HashMap<>();
+        props.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true");
+        props.put(TopicConfig.RETENTION_BYTES_CONFIG, String.valueOf(retentionBytes));
+        props.put(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(retentionMs));
+
+        props.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, String.valueOf(localRetentionMs));
+        props.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, String.valueOf(localRetentionBytes));
+        assertThrows(ConfigException.class,
+            () -> LogConfig.validate(Map.of(), props, kafkaConfig.extractLogConfigMap(),
+                kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+    }
+
+    @Test
+    public void testEnableRemoteLogStorageCleanupPolicy() {
+        KafkaConfig kafkaConfig = createKafkaConfig(true);
+        HashMap<String, String> logProps = new HashMap<>();
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE);
+        logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true");
+        LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT);
+        assertThrows(ConfigException.class, () -> LogConfig.validate(Map.of(), logProps,
+                kafkaConfig.extractLogConfigMap(), kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "delete,compact");
+        assertThrows(ConfigException.class, () -> LogConfig.validate(Map.of(), logProps,
+                kafkaConfig.extractLogConfigMap(), kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "compact,delete");
+        assertThrows(ConfigException.class, () -> LogConfig.validate(Map.of(), logProps,
+                kafkaConfig.extractLogConfigMap(), kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "delete,delete,delete");
+        LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+
+        logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "");
+        LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+    }
+
+    @ParameterizedTest(name = "testEnableRemoteLogStorage with sysRemoteStorageEnabled: {0}")
+    @ValueSource(booleans = {true, false})
+    public void testEnableRemoteLogStorage(boolean sysRemoteStorageEnabled) {
+        KafkaConfig kafkaConfig = createKafkaConfig(sysRemoteStorageEnabled);
+
+        HashMap<String, String> logProps = new HashMap<>();
+        logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true");
+        if (sysRemoteStorageEnabled) {
+            LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                    kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+        } else {
+            ConfigException message = assertThrows(ConfigException.class,
+                    () -> LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                            kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+            assertTrue(message.getMessage().contains("Tiered Storage functionality is disabled in the broker"));
+        }
+    }
+
+    @ParameterizedTest(name = "testDisableRemoteLogStorage with wasRemoteStorageEnabled: {0}")
+    @ValueSource(booleans = {true, false})
+    public void testDisableRemoteLogStorage(boolean wasRemoteStorageEnabled) {
+        KafkaConfig kafkaConfig = createKafkaConfig(true);
+
+        HashMap<String, String> logProps = new HashMap<>();
+        logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "false");
+        if (wasRemoteStorageEnabled) {
+            InvalidConfigurationException message = assertThrows(InvalidConfigurationException.class,
+                    () -> LogConfig.validate(Map.of(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true"),
+                            logProps, kafkaConfig.extractLogConfigMap(),
+                            kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+            assertTrue(message.getMessage().contains("It is invalid to disable remote storage without deleting remote data. " +
+                    "If you want to keep the remote data and turn to read only, please set `remote.storage.enable=true,remote.log.copy.disable=true`. " +
+                    "If you want to disable remote storage and delete all remote data, please set `remote.storage.enable=false,remote.log.delete.on.disable=true`."));
+
+            // It should be able to disable the remote log storage when delete on disable is set to true
+            logProps.put(TopicConfig.REMOTE_LOG_DELETE_ON_DISABLE_CONFIG, "true");
+            LogConfig.validate(Map.of(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true"),
+                    logProps, kafkaConfig.extractLogConfigMap(),
+                    kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+        } else {
+            LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                    kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+            LogConfig.validate(Map.of(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "false"), logProps,
+                    kafkaConfig.extractLogConfigMap(), kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+        }
+    }
+
+    @ParameterizedTest(name = "testTopicCreationWithInvalidRetentionTime with sysRemoteStorageEnabled: {0}")
+    @ValueSource(booleans = {true, false})
+    public void testTopicCreationWithInvalidRetentionTime(boolean sysRemoteStorageEnabled) {
+        Properties extra = new Properties();
+        extra.put(ServerLogConfigs.LOG_RETENTION_TIME_MILLIS_CONFIG, "1000");
+        extra.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, "900");
+        KafkaConfig kafkaConfig = createKafkaConfig(sysRemoteStorageEnabled, extra);
+
+        // Topic local log retention time inherited from Broker is greater than the topic's complete log retention time
+        HashMap<String, String> logProps = new HashMap<>();
+        logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, String.valueOf(sysRemoteStorageEnabled));
+        logProps.put(TopicConfig.RETENTION_MS_CONFIG, "500");
+        if (sysRemoteStorageEnabled) {
+            ConfigException message = assertThrows(ConfigException.class,
+                    () -> LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                            kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled()));
+            assertTrue(message.getMessage().contains(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG));
+        } else {
+            LogConfig.validate(Map.of(), logProps, kafkaConfig.extractLogConfigMap(),
+                    kafkaConfig.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
+        }
+    }
+
+    @ParameterizedTest(name = "testTopicCreationWithInvalidRetentionSize with sysRemoteStorageEnabled: {0}")
+    @ValueSource(booleans = {true, false})
+    public void testTopicCreationWithInvalidRetentionSize(boolean sysRemoteStorageEnabled) {
+        Properties extra = new Properties();
+        extra.put(ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG, "1024");
+        extra.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, "512");
+        KafkaConfig kafkaConfig = createKafkaConfig(sysRemoteStorageEnabled, extra);
+
+        // Topic local retention size inherited from Broker is greater than the topic's complete log retention size
+        HashMap<String, String> logProps = new HashMap<>();
+        logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, String.valueOf(sysRemoteStorageEnabled));
+        logProps.put(TopicConfig.RETENTION_BYTES_CONFIG, "128");

Review Comment:
   ```suggestion
           Map<String, String> logProps = Map.of(
                   TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, String.valueOf(sysRemoteStorageEnabled),
                   TopicConfig.RETENTION_MS_CONFIG, "500"
           );
   ```



-- 
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-20388: Move LogConfigTest from core to storage [kafka]

["m1a2st (via GitHub)" <gi...@apache.org>]

m1a2st commented on PR #21980:
URL: https://github.com/apache/kafka/pull/21980#issuecomment-4244734409

   > maybe the better would be to wait until KafkaConfig is moved as you suggested? I assume there are issues tracking such movement right?
   
   Since the `KafkaConfig` migration is still blocked by `DynamicBrokerConfig`, if the maintainers are okay with temporarily relying on Scala code, I think it’s reasonable to proceed with migrating this test.
   
   FYI: https://issues.apache.org/jira/browse/KAFKA-15853 


-- 
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-20388: Move LogConfigTest from core to storage [kafka]

["see-quick (via GitHub)" <gi...@apache.org>]

see-quick commented on code in PR #21980:
URL: https://github.com/apache/kafka/pull/21980#discussion_r3073308930


##########
storage/src/test/java/org/apache/kafka/storage/internals/log/LogConfigTest.java:
##########
@@ -0,0 +1,460 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.storage.internals.log;
+
+import kafka.server.KafkaConfig;
+import kafka.utils.TestUtils;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.errors.InvalidConfigurationException;
+import org.apache.kafka.server.config.ServerLogConfigs;
+import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class LogConfigTest {
+
+    @Test
+    public void testKafkaConfigToProps() {
+        long millisInHour = 60L * 60L * 1000L;
+        long millisInDay = 24L * millisInHour;
+        long bytesInGB = 1024L * 1024L * 1024L;
+        Properties kafkaProps = TestUtils.createDummyBrokerConfig();
+        kafkaProps.put(ServerLogConfigs.LOG_ROLL_TIME_HOURS_CONFIG, "2");
+        kafkaProps.put(ServerLogConfigs.LOG_ROLL_TIME_JITTER_HOURS_CONFIG, "2");
+        kafkaProps.put(ServerLogConfigs.LOG_RETENTION_TIME_HOURS_CONFIG, "960"); // 40 days
+        kafkaProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, "2592000000"); // 30 days
+        kafkaProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, "4294967296"); // 4 GB
+
+        Map<String, Object> logProps = KafkaConfig.fromProps(kafkaProps).extractLogConfigMap();
+        assertEquals(2 * millisInHour, logProps.get(TopicConfig.SEGMENT_MS_CONFIG));
+        assertEquals(2 * millisInHour, logProps.get(TopicConfig.SEGMENT_JITTER_MS_CONFIG));
+        assertEquals(40 * millisInDay, logProps.get(TopicConfig.RETENTION_MS_CONFIG));
+        assertEquals(30 * millisInDay, logProps.get(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG));
+        assertEquals(4 * bytesInGB, logProps.get(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG));
+    }
+
+    @Test
+    public void testFromPropsInvalid() {
+        for (String name : LogConfig.configNames()) {
+            Object[] invalidValues = invalidValuesForProp(name);
+            if (invalidValues != null) {
+                assertPropertyInvalid(name, invalidValues);
+            }
+        }
+    }
+
+    private Object[] invalidValuesForProp(String name) {
+        return switch (name) {
+            case TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG -> new Object[]{"not a boolean"};
+            case TopicConfig.RETENTION_BYTES_CONFIG, TopicConfig.RETENTION_MS_CONFIG -> new Object[]{"not_a_number"};
+            case TopicConfig.CLEANUP_POLICY_CONFIG -> new Object[]{"true", "foobar"};
+            case TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG -> new Object[]{"not_a_number", "-0.1", "1.2"};
+            case TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG -> new Object[]{"not_a_number", "0", "-1"};
+            case TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG -> new Object[]{"not_a_number", "-3"};
+            case TopicConfig.COMPRESSION_GZIP_LEVEL_CONFIG -> new Object[]{"not_a_number", "-2"};
+            case TopicConfig.COMPRESSION_LZ4_LEVEL_CONFIG -> new Object[]{"not_a_number", "-1"};
+            case TopicConfig.COMPRESSION_ZSTD_LEVEL_CONFIG -> new Object[]{"not_a_number", "-0.1"};
+            case TopicConfig.REMOTE_LOG_COPY_DISABLE_CONFIG, TopicConfig.REMOTE_LOG_DELETE_ON_DISABLE_CONFIG -> new Object[]{"not_a_number", "remove", "0"};
+            case LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG -> null;
+            default -> new Object[]{"not_a_number", "-1"};
+        };
+    }
+
+    @Test
+    public void testInvalidCompactionLagConfig() {
+        HashMap<String, String> props = new HashMap<>();
+        props.put(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG, "100");
+        props.put(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG, "200");
+        assertThrows(Exception.class, () -> LogConfig.validate(props));
+    }
+
+    @Test
+    public void shouldValidateThrottledReplicasConfig() {
+        assertTrue(isValid("*"));
+        assertTrue(isValid("* "));
+        assertTrue(isValid(""));
+        assertTrue(isValid(" "));
+        assertTrue(isValid("100:10"));
+        assertTrue(isValid("100:10,12:10"));
+        assertTrue(isValid("100:10,12:10,15:1"));
+        assertTrue(isValid("100:10,12:10,15:1  "));
+        assertTrue(isValid("100:0,"));
+
+        assertFalse(isValid("100"));
+        assertFalse(isValid("100:"));
+        assertFalse(isValid("100:0,10"));
+        assertFalse(isValid("100:0,10:"));
+        assertFalse(isValid("100:0,10:   "));
+        assertFalse(isValid("100 :0,10:   "));
+        assertFalse(isValid("100: 0,10:   "));
+        assertFalse(isValid("100:0,10 :   "));
+        assertFalse(isValid("*,100:10"));
+        assertFalse(isValid("* ,100:10"));
+    }
+
+    /* Sanity check that toHtmlTable produces one of the expected configs */
+    @Test
+    public void testToHtmlTable() {
+        String html = LogConfig.configDefCopy().toHtmlTable();
+        String expectedConfig = "<td>file.delete.delay.ms</td>";
+        assertTrue(html.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + html);
+    }
+
+    /* Sanity check that toHtml produces one of the expected configs */
+    @Test
+    public void testToHtml() {
+        String html = LogConfig.configDefCopy().toHtml(4, key -> "prefix_" + key, Map.of());
+        String expectedConfig = "<h4><a id=\"file.delete.delay.ms\"></a><a id=\"prefix_file.delete.delay.ms\" href=\"#prefix_file.delete.delay.ms\">file.delete.delay.ms</a></h4>";
+        assertTrue(html.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + html);
+    }
+
+    /* Sanity check that toEnrichedRst produces one of the expected configs */
+    @Test
+    public void testToEnrichedRst() {
+        String rst = LogConfig.configDefCopy().toEnrichedRst();
+        String expectedConfig = "``file.delete.delay.ms``";
+        assertTrue(rst.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + rst);
+    }
+
+    /* Sanity check that toRst produces one of the expected configs */
+    @Test
+    public void testToRst() {
+        String rst = LogConfig.configDefCopy().toRst();
+        String expectedConfig = "``file.delete.delay.ms``";
+        assertTrue(rst.contains(expectedConfig), "Could not find `" + expectedConfig + "` in:\n " + rst);
+    }
+
+    @Test
+    public void testGetConfigValue() {
+        // Add a config that doesn't set the `serverDefaultConfigName`
+        LogConfig.LogConfigDef configDef = LogConfig.configDefCopy();
+        String configNameWithNoServerMapping = "log.foo";
+        configDef.define(configNameWithNoServerMapping, ConfigDef.Type.INT, 1, ConfigDef.Importance.MEDIUM,
+                configNameWithNoServerMapping + " doc");
+
+        ConfigDef.ConfigKey deleteDelayKey = configDef.configKeys().get(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG);
+        Object deleteDelayServerDefault = configDef.getConfigValue(deleteDelayKey, LogConfig.SERVER_DEFAULT_HEADER_NAME);
+        assertEquals(ServerLogConfigs.LOG_DELETE_DELAY_MS_CONFIG, deleteDelayServerDefault);
+
+        ConfigDef.ConfigKey keyWithNoServerMapping = configDef.configKeys().get(configNameWithNoServerMapping);
+        Object nullServerDefault = configDef.getConfigValue(keyWithNoServerMapping, LogConfig.SERVER_DEFAULT_HEADER_NAME);
+        assertNull(nullServerDefault);
+    }
+
+    @Test
+    public void testOverriddenConfigsAsLoggableString() {
+        Properties kafkaProps = TestUtils.createDummyBrokerConfig();
+        kafkaProps.put("unknown.broker.password.config", "aaaaa");
+        kafkaProps.put(ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG, "50");
+        kafkaProps.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, "somekeypassword");
+        KafkaConfig kafkaConfig = KafkaConfig.fromProps(kafkaProps);
+        Properties topicOverrides = new Properties();
+        // Only set as a topic config
+        topicOverrides.setProperty(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2");
+        // Overrides value from broker config
+        topicOverrides.setProperty(TopicConfig.RETENTION_BYTES_CONFIG, "100");
+        // Unknown topic config, but known broker config
+        topicOverrides.setProperty(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "sometrustpasswrd");
+        // Unknown config
+        topicOverrides.setProperty("unknown.topic.password.config", "bbbb");
+        // We don't currently have any sensitive topic configs, if we add them, we should set one here
+        LogConfig logConfig = LogConfig.fromProps(kafkaConfig.extractLogConfigMap(), topicOverrides);
+        assertEquals("{min.insync.replicas=2, retention.bytes=100, ssl.truststore.password=(redacted), unknown.topic.password.config=(redacted)}",
+                logConfig.overriddenConfigsAsLoggableString());
+    }
+
+    private boolean isValid(String configValue) {
+        try {
+            ThrottledReplicaListValidator.ensureValidString("", configValue);
+            return true;
+        } catch (ConfigException e) {
+            return false;
+        }
+    }
+
+    private KafkaConfig createKafkaConfig(boolean remoteStorageEnabled, Properties extra) {
+        Properties props = TestUtils.createDummyBrokerConfig();

Review Comment:
   I think I saw this inside `RemoteLogManagerTest`. But in the end then we should create some dummy helper method in the storage? 



-- 
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-20388: Move LogConfigTest from core to storage [kafka]

["mimaison (via GitHub)" <gi...@apache.org>]

mimaison commented on PR #21980:
URL: https://github.com/apache/kafka/pull/21980#issuecomment-4496825701

   The test failures seem related:
   ```
   LogConfigTest > testTopicCreationWithInvalidRetentionSize with sysRemoteStorageEnabled: true
   ```
   failed with both Java 17 and 25


-- 
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