MINOR: move tiered storage related configs to a separate class within LogConfig (#11110)

The original code uses a RemoteLogManagerConfig class to store KIP-405 configs and adds three configs to LogConfig. This makes the code complicated and developers may be confused.

This PR allows us to access RemoteLogManagerConfig from KafkaConfig and do the same for LogConfig. Kafka developers will see the same interface for the KIP-405 configs. After this change, if we want to read remoteStorageEnable we should use LogConfig.tieredLogConfig.remoteStorageEnable instead of LogConfig.remoteStorageEnable. The same for localRetentionMs and localRetentionBytes. If we want to read configs in RemoteLogManagerConfig, we should use KafkaConfig.tieredKafkaConfig.xxx.

Reviewers: Satish Duggana <satishd@apache.org>, Kowshik Prakasam <kprakasam@confluent.io>, Jun Rao <junrao@gmail.com>

core/src/main/scala/kafka/log/LogConfig.scala

@@ -107,6 +107,8 @@ case class LogConfig(props: java.util.Map[_, _], overriddenConfigs: Set[String]
   val LeaderReplicationThrottledReplicas = getList(LogConfig.LeaderReplicationThrottledReplicasProp)
   val FollowerReplicationThrottledReplicas = getList(LogConfig.FollowerReplicationThrottledReplicasProp)
   val messageDownConversionEnable = getBoolean(LogConfig.MessageDownConversionEnableProp)
+
+  class RemoteLogConfig {
     val remoteStorageEnable = getBoolean(LogConfig.RemoteLogStorageEnableProp)
 
     val localRetentionMs: Long = {
@@ -132,7 +134,7 @@ case class LogConfig(props: java.util.Map[_, _], overriddenConfigs: Set[String]
       val localLogRetentionBytes = getLong(LogConfig.LocalLogRetentionBytesProp)
 
       // -2 indicates to derive value from retentionSize property.
-    if(localLogRetentionBytes == -2) retentionSize;
+      if(localLogRetentionBytes == -2) retentionSize
       else {
         // Added validation here to check the effective value should not be more than RetentionBytes.
         if(localLogRetentionBytes == -1 && retentionSize != -1) {
@@ -146,6 +148,10 @@ case class LogConfig(props: java.util.Map[_, _], overriddenConfigs: Set[String]
         localLogRetentionBytes
       }
     }
+  }
+
+  private val _remoteLogConfig = new RemoteLogConfig()
+  def remoteLogConfig = _remoteLogConfig
 
   @nowarn("cat=deprecation")
   def recordVersion = messageFormatVersion.recordVersion

core/src/main/scala/kafka/server/KafkaConfig.scala

@@ -1435,6 +1435,9 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
   val zkEnableSecureAcls: Boolean = getBoolean(KafkaConfig.ZkEnableSecureAclsProp)
   val zkMaxInFlightRequests: Int = getInt(KafkaConfig.ZkMaxInFlightRequestsProp)
 
+  private val _remoteLogManagerConfig = new RemoteLogManagerConfig(this)
+  def remoteLogManagerConfig = _remoteLogManagerConfig
+
   private def zkBooleanConfigOrSystemPropertyWithDefaultValue(propKey: String): Boolean = {
     // Use the system property if it exists and the Kafka config value was defaulted rather than actually provided
     // Need to translate any system property value from true/false (String) to true/false (Boolean)

core/src/test/scala/unit/kafka/log/LogConfigTest.scala

@@ -222,8 +222,8 @@ class LogConfigTest {
     props.put(LogConfig.RetentionMsProp, retentionMs.toString)
     val logConfig = new LogConfig(props)
 
-    assertEquals(retentionMs, logConfig.localRetentionMs)
-    assertEquals(retentionBytes, logConfig.localRetentionBytes)
+    assertEquals(retentionMs, logConfig.remoteLogConfig.localRetentionMs)
+    assertEquals(retentionBytes, logConfig.remoteLogConfig.localRetentionBytes)
   }
 
   @Test
@@ -231,8 +231,8 @@ class LogConfigTest {
     val logConfig = new LogConfig( new Properties())
 
     // Local retention defaults are derived from retention properties which can be default or custom.
-    assertEquals(Defaults.RetentionMs, logConfig.localRetentionMs)
-    assertEquals(Defaults.RetentionSize, logConfig.localRetentionBytes)
+    assertEquals(Defaults.RetentionMs, logConfig.remoteLogConfig.localRetentionMs)
+    assertEquals(Defaults.RetentionSize, logConfig.remoteLogConfig.localRetentionBytes)
   }
 
   @Test
@@ -247,8 +247,8 @@ class LogConfigTest {
     props.put(LogConfig.LocalLogRetentionBytesProp, localRetentionBytes.toString)
     val logConfig = new LogConfig(props)
 
-    assertEquals(localRetentionMs, logConfig.localRetentionMs)
-    assertEquals(localRetentionBytes, logConfig.localRetentionBytes)
+    assertEquals(localRetentionMs, logConfig.remoteLogConfig.localRetentionMs)
+    assertEquals(localRetentionBytes, logConfig.remoteLogConfig.localRetentionBytes)
   }
 
   @Test

storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java

@@ -253,9 +253,13 @@ public final class RemoteLogManagerConfig {
              config.getInt(REMOTE_LOG_READER_THREADS_PROP),
              config.getInt(REMOTE_LOG_READER_MAX_PENDING_TASKS_PROP),
              config.getString(REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP),
-             config.originalsWithPrefix(config.getString(REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP)),
+             config.getString(REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP) != null
+                 ? config.originalsWithPrefix(config.getString(REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP))
+                 : Collections.emptyMap(),
              config.getString(REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP),
-             config.originalsWithPrefix(config.getString(REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP)));
+             config.getString(REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP) != null
+                 ? config.originalsWithPrefix(config.getString(REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP))
+                 : Collections.emptyMap());
     }
 
     // Visible for testing